Adds cleanup after shutdown of plugin

Fixes #10691

logstash-core/lib/logstash/plugin.rb

@@ -72,7 +72,11 @@ class LogStash::Plugin
   # main task terminates
   def do_close
     @logger.debug("Closing", :plugin => self.class.name)
-    close
+    begin
+      close
+    ensure
+      LogStash::PluginMetadata.delete_for_plugin(self.id)
+    end
   end
 
   # Subclasses should implement this close method if you need to perform any

logstash-core/lib/logstash/plugin_metadata.rb

@@ -7,8 +7,7 @@ module LogStash
   # `PluginMetadata` provides a space to store key/value metadata about a plugin, typically metadata about
   # external resources that can be gleaned during plugin registration.
   #
-  # Data is persisted across pipeline reloads, and no effort is made to clean up metadata from pipelines
-  # that no longer exist after a pipeline reload.
+  # Data should not be persisted across pipeline reloads, and should be cleaned up after a pipeline reload
   #
   #  - It MUST NOT be used to store processing state
   #  - It SHOULD NOT be updated frequently.
@@ -28,6 +27,7 @@ module LogStash
   #
   # @since 7.1
   class PluginMetadata
+    include LogStash::Util::Loggable
 
     Thread.exclusive do
       @registry = ThreadSafe::Cache.new unless defined?(@registry)
@@ -63,6 +63,7 @@ module LogStash
       #
       # @return [Boolean]
       def delete_for_plugin(plugin_id)
+        logger.debug("Removing metadata for plugin #{plugin_id}")
         old_registry = @registry.delete(plugin_id)
         old_registry.clear unless old_registry.nil?
       end

logstash-core/spec/logstash/plugin_spec.rb

@@ -328,6 +328,14 @@ describe LogStash::Plugin do
               expect(old_value).to be_nil
               expect(plugin_instance.plugin_metadata.get(:foo)).to eq(new_value)
             end
+
+            it 'removes metadata when the plugin is closed' do
+              new_value = 'foo'
+              plugin_instance.plugin_metadata.set(:foo, new_value)
+              expect(plugin_instance.plugin_metadata.get(:foo)).to eq(new_value)
+              plugin_instance.do_close
+              expect(plugin_instance.plugin_metadata.set?(:foo)).to be_falsey
+            end
           end
         end
       end