Hi I’m trying to run logstash with input as Kafka and output to file, here is the pipeline config file:
input {
kafka
{
bootstrap_servers => "kafka_server:9092
topics => ["Topic1", "Topic2", "Topic3"]
codec => "json"
group_id => "logstash-consumers-1"
consumer_threads => 1
decorate_events => true
auto_offset_reset => "earliest"
}
}
output{
file{
path => "/home/avs/output4.out"
}
}
The output for --config.test_and_exit is:
[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 {"logstash.version"=>"7.16.3", "jruby.version"=>"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]"}
[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
But getting an error while trying to execute the pipeline from the command line:
[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 {"logstash.version"=>"7.16.3", "jruby.version"=>"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]"}
[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=>9600, :ssl_enabled=>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=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>500, "pipeline.sources"=>["/opt/oss/logstash-7.16.3/config/pipelines/ipv4.conf"], :thread=>"#<Thread:0x1fad61da run>"}
[2022-01-24T02:59:20,903][INFO ][logstash.javapipeline ][main] Pipeline Java execution initialization time {"seconds"=>1.4}
[2022-01-24T02:59:20,962][INFO ][logstash.javapipeline ][main] Pipeline started {"pipeline.id"=>"main"}
[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=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[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 {"pipeline.id"=>"main"}
[2022-01-24T02:59:23,422][FATAL][org.logstash.Logstash ]
org.jruby.exceptions.ThreadKill: null
The Kafka server is running on AWS and port 9092 is reachable, verified using telnet:
[avs@localhost ~]$ telnet kafka_host 9092
Trying xx.x.xx.xxx...
Connected to kafka_host.
Escape character is '^]'.