Fix/get host context (#93)

* Fix `getHostContext` method accessibility

This commit works around a recent change to jruby where calls to `java_send` to
package private accessible Java methods will fail with
`(NoMethodError) undefined method `getHostContext'` errors.

This commit works around this by using Java reflection to make the method accessible.

I have verified that this is the only package private method that we call on that class.

Fixes #91

Author: robbavey

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 1.4.9
+  - Fixed issue with `getHostContext` method accessibility, causing plugin not to be able to run  [#93](https://github.com/logstash-plugins/logstash-input-azure_event_hubs/pull/93)
+
 ## 1.4.8
   - Fixed connection placeholder replacements errors with Logstash `8.15.1` and `8.15.2` [#92](https://github.com/logstash-plugins/logstash-input-azure_event_hubs/pull/92)
 

VERSION

@@ -1 +1 @@
-1.4.8
+1.4.9

lib/logstash-input-azure_event_hubs.rb

@@ -5,5 +5,5 @@
 require_jar('com.microsoft.azure', 'qpid-proton-j-extensions', '1.2.4')
 require_jar('com.microsoft.azure', 'azure-eventhubs-eph', '2.5.2')
 require_jar('com.microsoft.azure', 'azure-storage', '8.6.6')
-require_jar('com.google.code.gson', 'gson', '2.8.5')
+require_jar('com.google.code.gson', 'gson', '2.10.1')
 require_jar('org.apache.qpid', 'proton-j', '0.33.9')

lib/logstash/inputs/azure_event_hubs.rb

@@ -411,20 +411,7 @@ def run(queue)
                 scheduled_executor_service)
           else
             @logger.warn("You have NOT specified a `storage_connection_string` for #{event_hub_name}. This configuration is only supported for a single Logstash instance.")
-            checkpoint_manager = InMemoryCheckpointManager.new
-            lease_manager = InMemoryLeaseManager.new
-            event_processor_host = EventProcessorHost.new(
-                EventProcessorHost.createHostName('logstash'),
-                event_hub_name,
-                event_hub['consumer_group'],
-                event_hub['event_hub_connections'].first.value, #there will only be one in this array by the time it gets here
-                checkpoint_manager,
-                lease_manager,
-                scheduled_executor_service,
-                nil)
-            #using java_send to avoid naming conflicts with 'initialize' method
-            lease_manager.java_send :initialize, [HostContext], event_processor_host.getHostContext
-            checkpoint_manager.java_send :initialize, [HostContext], event_processor_host.getHostContext
+            event_processor_host = create_in_memory_event_processor_host(event_hub, event_hub_name, scheduled_executor_service)
           end
           options = EventProcessorOptions.new
           options.setMaxBatchSize(max_batch_size)
@@ -496,4 +483,38 @@ def run(queue)
       @logger.debug("interrupted while waiting to close executor service, this can generally be ignored", :exception => e, :backtrace => e.backtrace) if e
     end
   end
+
+  def create_in_memory_event_processor_host(event_hub, event_hub_name, scheduled_executor_service)
+    checkpoint_manager = InMemoryCheckpointManager.new
+    lease_manager = InMemoryLeaseManager.new
+    event_processor_host = EventProcessorHost.new(
+      EventProcessorHost.createHostName('logstash'),
+      event_hub_name,
+      event_hub['consumer_group'],
+      event_hub['event_hub_connections'].first.value, #there will only be one in this array by the time it gets here
+      checkpoint_manager,
+      lease_manager,
+      scheduled_executor_service,
+      nil)
+    host_context = get_host_context(event_processor_host)
+    #using java_send to avoid naming conflicts with 'initialize' method
+    lease_manager.java_send :initialize, [HostContext], host_context
+    checkpoint_manager.java_send :initialize, [HostContext], host_context
+    event_processor_host
+  end
+
+  private
+
+  # This method is used to get around the fact that recent versions of jruby do not
+  # allow access to the package private protected method `getHostContext`
+  def get_host_context(event_processor_host)
+    call_private(event_processor_host, 'getHostContext')
+  end
+
+  def call_private(clazz, method)
+    method = clazz.java_class.declared_method(method)
+    method.accessible = true
+    method.invoke(clazz)
+  end
+
 end

spec/inputs/azure_event_hub_spec.rb

@@ -207,10 +207,10 @@
         register_counter = java.util.concurrent.atomic.AtomicInteger.new
         unregister_counter = java.util.concurrent.atomic.AtomicInteger.new
         assertion_count = java.util.concurrent.atomic.AtomicInteger.new
+        allow(input).to receive(:get_host_context) {mock_host_context}
         allow_any_instance_of(InMemoryLeaseManager).to receive(:java_send)
         allow_any_instance_of(InMemoryCheckpointManager).to receive(:java_send)
 
-        allow(mock_host).to receive(:getHostContext) {mock_host_context}
         allow(mock_host_context).to receive(:getEventHubPath) {"foo"}
 
         expect(mock_host).to receive(:registerEventProcessorFactory).at_most(3).times {
@@ -256,6 +256,11 @@
         expect(assertion_count.get).to be == 3
       end
 
+      it "can create an in memory EPH" do
+        #event_hub, event_hub_name, scheduled_executor_service
+        exploded_config = input.event_hubs_exploded
+        input.create_in_memory_event_processor_host(exploded_config[0], exploded_config[0]['event_hubs'].first, nil)
+      end
     end
 
     describe "Bad Basic Config" do