title: Logstash error while connecting to kafka?
body: <p>Hi I'm trying to run logstash with input as Kafka and output to file, here is the pipeline config file:</p>
  <pre><code>input {
    kafka
    {
         bootstrap_servers =&gt; &quot;kafka_server:9092
         topics =&gt; [&quot;Topic1&quot;, &quot;Topic2&quot;, &quot;Topic3&quot;]
         codec =&gt; &quot;json&quot;
         group_id =&gt; &quot;logstash-consumers-1&quot;
         consumer_threads =&gt; 1
         decorate_events =&gt; true
         auto_offset_reset =&gt; &quot;earliest&quot;
    }
  }
  
  output{
         file{
             path =&gt; &quot;/home/avs/output4.out&quot;
         }
  }
  </code></pre>
  <p>The output for <code>--config.test_and_exit</code> is:</p>
  <pre><code>[avs@localhost logstash-7.16.3]$ ./bin/logstash --config.test_and_exit -f config/pipelines/ipv4.conf 
  Using bundled JDK: /opt/oss/logstash-7.16.3/jdk
  OpenJDK 64-Bit Server VM warning: Option UseConcMarkSweepGC was deprecated in version 9.0 and will likely be removed in a future release.
  Sending Logstash logs to /opt/oss/data/logs/logstash which is now configured via log4j2.properties
  [2022-01-24T02:57:58,420][INFO ][logstash.runner          ] Log4j configuration path used is: /opt/oss/logstash-7.16.3/config/log4j2.properties
  [2022-01-24T02:57:58,462][INFO ][logstash.runner          ] Starting Logstash {&quot;logstash.version&quot;=&gt;&quot;7.16.3&quot;, &quot;jruby.version&quot;=&gt;&quot;jruby 9.2.20.1 (2.5.8) 2021-11-30 2a2962fbd1 OpenJDK 64-Bit Server VM 11.0.13+8 on 11.0.13+8 +indy +jit [linux-x86_64]&quot;}
  [2022-01-24T02:57:59,178][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
  [2022-01-24T02:58:01,274][INFO ][org.reflections.Reflections] Reflections took 108 ms to scan 1 urls, producing 119 keys and 417 values 
  Configuration OK
  [2022-01-24T02:58:02,737][INFO ][logstash.runner          ] Using config.test_and_exit mode. Config Validation Result: OK. Exiting Logstash
  
  </code></pre>
  <p>But getting an error while trying to execute the pipeline from the command line:</p>
  <pre><code>[avs@localhost logstash-7.16.3]$ ./bin/logstash -f config/pipelines/ipv4.conf 
  Using bundled JDK: /opt/oss/logstash-7.16.3/jdk
  OpenJDK 64-Bit Server VM warning: Option UseConcMarkSweepGC was deprecated in version 9.0 and will likely be removed in a future release.
  Sending Logstash logs to /opt/oss/data/logs/logstash which is now configured via log4j2.properties
  [2022-01-24T02:59:11,806][INFO ][logstash.runner          ] Log4j configuration path used is: /opt/oss/logstash-7.16.3/config/log4j2.properties
  [2022-01-24T02:59:11,826][INFO ][logstash.runner          ] Starting Logstash {&quot;logstash.version&quot;=&gt;&quot;7.16.3&quot;, &quot;jruby.version&quot;=&gt;&quot;jruby 9.2.20.1 (2.5.8) 2021-11-30 2a2962fbd1 OpenJDK 64-Bit Server VM 11.0.13+8 on 11.0.13+8 +indy +jit [linux-x86_64]&quot;}
  [2022-01-24T02:59:12,412][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
  [2022-01-24T02:59:15,324][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=&gt;9600, :ssl_enabled=&gt;false}
  [2022-01-24T02:59:17,619][INFO ][org.reflections.Reflections] Reflections took 127 ms to scan 1 urls, producing 119 keys and 417 values 
  [2022-01-24T02:59:19,489][INFO ][logstash.javapipeline    ][main] Starting pipeline {:pipeline_id=&gt;&quot;main&quot;, &quot;pipeline.workers&quot;=&gt;4, &quot;pipeline.batch.size&quot;=&gt;125, &quot;pipeline.batch.delay&quot;=&gt;50, &quot;pipeline.max_inflight&quot;=&gt;500, &quot;pipeline.sources&quot;=&gt;[&quot;/opt/oss/logstash-7.16.3/config/pipelines/ipv4.conf&quot;], :thread=&gt;&quot;#&lt;Thread:0x1fad61da run&gt;&quot;}
  [2022-01-24T02:59:20,903][INFO ][logstash.javapipeline    ][main] Pipeline Java execution initialization time {&quot;seconds&quot;=&gt;1.4}
  [2022-01-24T02:59:20,962][INFO ][logstash.javapipeline    ][main] Pipeline started {&quot;pipeline.id&quot;=&gt;&quot;main&quot;}
  [2022-01-24T02:59:21,173][INFO ][org.apache.kafka.clients.consumer.ConsumerConfig][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] ConsumerConfig values: 
          allow.auto.create.topics = true
          auto.commit.interval.ms = 5000
          auto.offset.reset = earliest
          bootstrap.servers = [kafka_server:9092]
          check.crcs = true
          client.dns.lookup = default
          client.id = logstash-0
          client.rack = 
          connections.max.idle.ms = 540000
          default.api.timeout.ms = 60000
          enable.auto.commit = true
          exclude.internal.topics = true
          fetch.max.bytes = 52428800
          fetch.max.wait.ms = 500
          fetch.min.bytes = 1
          group.id = logstash-consumers-1
          group.instance.id = null
          heartbeat.interval.ms = 3000
          interceptor.classes = []
          internal.leave.group.on.close = true
          isolation.level = read_uncommitted
          key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
          max.partition.fetch.bytes = 1048576
          max.poll.interval.ms = 300000
          max.poll.records = 500
          metadata.max.age.ms = 300000
          metric.reporters = []
          metrics.num.samples = 2
          metrics.recording.level = INFO
          metrics.sample.window.ms = 30000
          partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
          receive.buffer.bytes = 32768
          reconnect.backoff.max.ms = 50
          reconnect.backoff.ms = 50
          request.timeout.ms = 40000
          retry.backoff.ms = 100
          sasl.client.callback.handler.class = null
          sasl.jaas.config = null
          sasl.kerberos.kinit.cmd = /usr/bin/kinit
          sasl.kerberos.min.time.before.relogin = 60000
          sasl.kerberos.service.name = null
          sasl.kerberos.ticket.renew.jitter = 0.05
          sasl.kerberos.ticket.renew.window.factor = 0.8
          sasl.login.callback.handler.class = null
          sasl.login.class = null
          sasl.login.refresh.buffer.seconds = 300
          sasl.login.refresh.min.period.seconds = 60
          sasl.login.refresh.window.factor = 0.8
          sasl.login.refresh.window.jitter = 0.05
          sasl.mechanism = GSSAPI
          security.protocol = PLAINTEXT
          security.providers = null
          send.buffer.bytes = 131072
          session.timeout.ms = 10000
          ssl.cipher.suites = null
          ssl.enabled.protocols = [TLSv1.2]
          ssl.endpoint.identification.algorithm = https
          ssl.key.password = null
          ssl.keymanager.algorithm = SunX509
          ssl.keystore.location = null
          ssl.keystore.password = null
          ssl.keystore.type = JKS
          ssl.protocol = TLSv1.2
          ssl.provider = null
          ssl.secure.random.implementation = null
          ssl.trustmanager.algorithm = PKIX
          ssl.truststore.location = null
          ssl.truststore.password = null
          ssl.truststore.type = JKS
          value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
  
  [2022-01-24T02:59:21,218][INFO ][logstash.agent           ] Pipelines running {:count=&gt;1, :running_pipelines=&gt;[:main], :non_running_pipelines=&gt;[]}
  [2022-01-24T02:59:21,407][INFO ][org.apache.kafka.common.utils.AppInfoParser][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] Kafka version: 2.5.1
  [2022-01-24T02:59:21,407][INFO ][org.apache.kafka.common.utils.AppInfoParser][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] Kafka commitId: 0efa8fb0f4c73d92
  [2022-01-24T02:59:21,407][INFO ][org.apache.kafka.common.utils.AppInfoParser][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] Kafka startTimeMs: 1643011161403
  [2022-01-24T02:59:21,420][INFO ][org.apache.kafka.clients.consumer.KafkaConsumer][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] [Consumer clientId=logstash-0, groupId=logstash-consumers-1] Subscribed to topic(s): ext_device-control-events_10121, ext_device-event_10121, ext_device-measurement_10121, ext_device-parameter_10121, ext_device-telemetry_10121, ext_device-topology_10121, ext_device_10121, ext_incoming-device-events_10121, ext_master-data-events_10121, ext_metering-point_10121, ext_process-events_10121, ext_register-statistic_10121, ext_device-process_10121, ext_device_measurement_10121, ext_device_measurment_10121, ext_deviceevent_10121, ext_mdus-events_10121, ext_operational-process_10121, ext_registerstatistic_10121
  [2022-01-24T02:59:22,129][INFO ][org.apache.kafka.clients.Metadata][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] [Consumer clientId=logstash-0, groupId=logstash-consumers-1] Cluster ID: KDpznPsAQyKSR2RDb8FCXQ
  [2022-01-24T02:59:22,132][INFO ][org.apache.kafka.clients.consumer.internals.AbstractCoordinator][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] [Consumer clientId=logstash-0, groupId=logstash-consumers-1] Discovered group coordinator ip-10-92-4-180.ap-south-1.compute.internal:kafka_port (id: 2147483647 rack: null)
  [2022-01-24T02:59:22,200][WARN ][org.apache.kafka.clients.NetworkClient][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] [Consumer clientId=logstash-0, groupId=logstash-consumers-1] Error connecting to node ip-10-92-4-180.ap-south-1.compute.internal:kafka_port (id: 2147483647 rack: null)
  java.net.UnknownHostException: ip-10-92-4-180.ap-south-1.compute.internal: Name or service not known
          at java.net.Inet6AddressImpl.lookupAllHostAddr(Native Method) ~[?:?]
          at java.net.InetAddress$PlatformNameService.lookupAllHostAddr(InetAddress.java:929) ~[?:?]
          at java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1519) ~[?:?]
          at java.net.InetAddress$NameServiceAddresses.get(InetAddress.java:848) ~[?:?]
          at java.net.InetAddress.getAllByName0(InetAddress.java:1509) ~[?:?]
          at java.net.InetAddress.getAllByName(InetAddress.java:1368) ~[?:?]
          at java.net.InetAddress.getAllByName(InetAddress.java:1302) ~[?:?]
          at org.apache.kafka.clients.ClientUtils.resolve(ClientUtils.java:110) ~[kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.currentAddress(ClusterConnectionStates.java:403) ~[kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.access$200(ClusterConnectionStates.java:363) ~[kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.ClusterConnectionStates.currentAddress(ClusterConnectionStates.java:151) ~[kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:962) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:294) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.tryConnect(ConsumerNetworkClient.java:572) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$FindCoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:798) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$FindCoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:778) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:204) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:167) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:127) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:599) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(ConsumerNetworkClient.java:409) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:294) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:236) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:469) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1274) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1238) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1173) [kafka-clients-2.5.1.jar:?]
          at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
          at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:?]
          at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:?]
          at java.lang.reflect.Method.invoke(Method.java:566) ~[?:?]
          at org.jruby.javasupport.JavaMethod.invokeDirectWithExceptionHandling(JavaMethod.java:426) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.javasupport.JavaMethod.invokeDirect(JavaMethod.java:293) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.java.invokers.InstanceMethodInvoker.call(InstanceMethodInvoker.java:24) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.java.invokers.InstanceMethodInvoker.call(InstanceMethodInvoker.java:86) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.ir.targets.InvokeSite.invoke(InvokeSite.java:207) [jruby-complete-9.2.20.1.jar:?]
          at opt.oss.logstash_minus_7_dot_16_dot_3.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_kafka_minus_10_dot_8_dot_2_minus_java.lib.logstash.inputs.kafka.RUBY$method$do_poll$0(/opt/oss/logstash-7.16.3/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.8.2-java/lib/logstash/inputs/kafka.rb:328) [jruby-complete-9.2.20.1.jar:?]
          at opt.oss.logstash_minus_7_dot_16_dot_3.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_kafka_minus_10_dot_8_dot_2_minus_java.lib.logstash.inputs.kafka.RUBY$method$do_poll$0$__VARARGS__(/opt/oss/logstash-7.16.3/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.8.2-java/lib/logstash/inputs/kafka.rb:325) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:80) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:70) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.ir.targets.InvokeSite.invoke(InvokeSite.java:207) [jruby-complete-9.2.20.1.jar:?]
          at opt.oss.logstash_minus_7_dot_16_dot_3.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_kafka_minus_10_dot_8_dot_2_minus_java.lib.logstash.inputs.kafka.RUBY$block$thread_runner$1(/opt/oss/logstash-7.16.3/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.8.2-java/lib/logstash/inputs/kafka.rb:313) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.runtime.CompiledIRBlockBody.callDirect(CompiledIRBlockBody.java:138) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:58) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:52) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.runtime.Block.call(Block.java:139) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.RubyProc.call(RubyProc.java:318) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.javasupport.Java$ProcToInterface.callProc(Java.java:1136) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.javasupport.Java$ProcToInterface.access$300(Java.java:1113) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.javasupport.Java$ProcToInterface$ConcreteMethod.call(Java.java:1174) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.gen.InterfaceImpl1948101582.run(org/jruby/gen/InterfaceImpl1948101582.gen:13) [jruby-complete-9.2.20.1.jar:?]
          at java.lang.Thread.run(Thread.java:829) [?:?]
  [2022-01-24T02:59:22,225][INFO ][org.apache.kafka.clients.consumer.internals.AbstractCoordinator][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] [Consumer clientId=logstash-0, groupId=logstash-consumers-1] (Re-)joining group
  [2022-01-24T02:59:22,234][INFO ][org.apache.kafka.clients.consumer.internals.AbstractCoordinator][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] [Consumer clientId=logstash-0, groupId=logstash-consumers-1] Group coordinator ip-10-92-4-180.ap-south-1.compute.internal:kafka_port (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
  [2022-01-24T02:59:22,234][INFO ][org.apache.kafka.clients.consumer.internals.AbstractCoordinator][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] [Consumer clientId=logstash-0, groupId=logstash-consumers-1] Join group failed with org.apache.kafka.common.errors.DisconnectException
  [2022-01-24T02:59:22,338][WARN ][org.apache.kafka.clients.NetworkClient][main][e00e0156d10b7702327422edaa4ed7878321fcdad61181e357c9bff93ca75395] [Consumer clientId=logstash-0, groupId=logstash-consumers-1] Error connecting to node ip-10-92-4-180.ap-south-1.compute.internal:kafka_port (id: 0 rack: null)
  java.net.UnknownHostException: ip-10-92-4-180.ap-south-1.compute.internal
          at java.net.InetAddress$CachedAddresses.get(InetAddress.java:797) ~[?:?]
          at java.net.InetAddress.getAllByName0(InetAddress.java:1509) ~[?:?]
          at java.net.InetAddress.getAllByName(InetAddress.java:1368) ~[?:?]
          at java.net.InetAddress.getAllByName(InetAddress.java:1302) ~[?:?]
          at org.apache.kafka.clients.ClientUtils.resolve(ClientUtils.java:110) ~[kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.currentAddress(ClusterConnectionStates.java:403) ~[kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.access$200(ClusterConnectionStates.java:363) ~[kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.ClusterConnectionStates.currentAddress(ClusterConnectionStates.java:151) ~[kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:962) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:294) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.trySend(ConsumerNetworkClient.java:495) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:252) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:236) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:401) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:358) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:497) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1274) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1238) [kafka-clients-2.5.1.jar:?]
          at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1173) [kafka-clients-2.5.1.jar:?]
          at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
          at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:?]
          at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:?]
          at java.lang.reflect.Method.invoke(Method.java:566) ~[?:?]
          at org.jruby.javasupport.JavaMethod.invokeDirectWithExceptionHandling(JavaMethod.java:426) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.javasupport.JavaMethod.invokeDirect(JavaMethod.java:293) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.java.invokers.InstanceMethodInvoker.call(InstanceMethodInvoker.java:24) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.java.invokers.InstanceMethodInvoker.call(InstanceMethodInvoker.java:86) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.ir.targets.InvokeSite.invoke(InvokeSite.java:207) [jruby-complete-9.2.20.1.jar:?]
          at opt.oss.logstash_minus_7_dot_16_dot_3.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_kafka_minus_10_dot_8_dot_2_minus_java.lib.logstash.inputs.kafka.RUBY$method$do_poll$0(/opt/oss/logstash-7.16.3/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.8.2-java/lib/logstash/inputs/kafka.rb:328) [jruby-complete-9.2.20.1.jar:?]
          at opt.oss.logstash_minus_7_dot_16_dot_3.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_kafka_minus_10_dot_8_dot_2_minus_java.lib.logstash.inputs.kafka.RUBY$method$do_poll$0$__VARARGS__(/opt/oss/logstash-7.16.3/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.8.2-java/lib/logstash/inputs/kafka.rb:325) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:80) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:70) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.ir.targets.InvokeSite.invoke(InvokeSite.java:207) [jruby-complete-9.2.20.1.jar:?]
          at opt.oss.logstash_minus_7_dot_16_dot_3.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_kafka_minus_10_dot_8_dot_2_minus_java.lib.logstash.inputs.kafka.RUBY$block$thread_runner$1(/opt/oss/logstash-7.16.3/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.8.2-java/lib/logstash/inputs/kafka.rb:313) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.runtime.CompiledIRBlockBody.callDirect(CompiledIRBlockBody.java:138) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:58) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:52) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.runtime.Block.call(Block.java:139) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.RubyProc.call(RubyProc.java:318) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.javasupport.Java$ProcToInterface.callProc(Java.java:1136) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.javasupport.Java$ProcToInterface.access$300(Java.java:1113) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.javasupport.Java$ProcToInterface$ConcreteMethod.call(Java.java:1174) [jruby-complete-9.2.20.1.jar:?]
          at org.jruby.gen.InterfaceImpl1948101582.run(org/jruby/gen/InterfaceImpl1948101582.gen:13) [jruby-complete-9.2.20.1.jar:?]
          at java.lang.Thread.run(Thread.java:829) [?:?]
  ^C[2022-01-24T02:59:23,259][FATAL][logstash.runner          ] SIGINT received. Terminating immediately..
  [2022-01-24T02:59:23,370][INFO ][logstash.javapipeline    ][main] Pipeline terminated {&quot;pipeline.id&quot;=&gt;&quot;main&quot;}
  [2022-01-24T02:59:23,422][FATAL][org.logstash.Logstash    ] 
  org.jruby.exceptions.ThreadKill: null
  </code></pre>
  <p>The Kafka server is running on AWS and port 9092 is reachable, verified using telnet:</p>
  <pre><code>[avs@localhost ~]$ telnet kafka_host 9092
  Trying xx.x.xx.xxx...
  Connected to kafka_host.
  Escape character is '^]'.
  </code></pre>