Kafka output gets permanently stuck after temporary wrong port — pipeline-to-pipeline address unavailable & retrying_send stall

Elastic Stack Logstash
1

Kafka output gets permanently stuck after temporary wrong port — pipeline-to-pipeline address unavailable & retrying_send stall

Hi,
I am facing an issue with Logstash Kafka output when the Kafka broker becomes temporarily unreachable.

Error :
[2025-12-05T12:41:32,869][WARN ][org.logstash.plugins.pipeline.AbstractPipelineBus][6c439cc070c5500b3acef737847adbb5978f7d66189e9c4591ef93fa3f24f09f] Attempted to send event to 'forward_kafka but that address was unavailable. Maybe the destination pipeline is down or stopping? Will Retry.

Problem Scenario

  • I intentionally changed the Kafka port to a wrong port (ex: 7879).
  • Logstash started showing Kafka connection errors (expected).
  • After a few seconds, I restored the correct port 7878 again.
  • Even after restoring the correct port, Logstash never recovers and it stops forwarding logs permanently.
  • Restarting Logstash fixes it, but I want Logstash to automatically recover without restart.

My Logstash pipeline forwards logs to Kafka using this output configuration:

output {
  kafka {
    codec => json
    topic_id => "temp-logs"
    bootstrap_servers => "dev.custom.com:7878"

    client_id => "DEV-00135"
    security_protocol => "SASL_PLAINTEXT"

    sasl_mechanism => "PLAIN"
    sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='admin-secret';"
  }
}


How to resolve this issue
2

If you like to reload configuration changes in .conf automatically, without LS restart, you can set in logstash.yml:

config.reload.automatic: true
config.reload.interval: 10s

You can set reconnect values in the kafka output params:

  • retries
  • retry_backoff_ms
  • reconnect_backoff_max_ms
  • reconnect_backoff_ms
  • request_timeout_ms

You can check default values, and do fine tuning for your case.
The kafka plugin v 10.50 is from April 2020, not sure which LS version do you have.

3

@Rios

config.reload.automatic: true
config.reload.interval: 10s

I already have these values in logstash.yml

I have Logstash v 8.19.3

4

@Rios

retries => 10
retry_backoff_ms => 500
reconnect_backoff_ms => 500
request_timeout_ms => 30000

I have added these to my config but also the same issue persists

Shoes error like

image
image1850×552 80.4 KB

5

What exactly you did and how are your pipelines connected? You mentioned that you are you using pipeline-to-pipeline communicaton, but you didn't share any of your configurations.

Please share your pipelines.yml file and the inputs and ouputs of all your pipelines running on the same instance.

6

@leandrojmp

  - pipeline.id: main_input
    path.config: "config/main_input.conf"
  - pipeline.id: input_snmp_trap
    path.config: "config/input_snmp_trap.conf"
  - pipeline.id: forward_syslog
    path.config: "config/forward_syslog.conf"
  - pipeline.id: input_eventlog
    path.config: "config/input_eventlog.conf"
    queue.max_bytes: 2048mb
    queue.type: persisted
  - pipeline.id: forward_eventlog
    path.config: "config/forward_eventlog.conf"
    queue.max_bytes: 1024mb
    queue.type: persisted
  - pipeline.id: forward_kafka
    path.config: "config/forward_kafka.conf"
    queue.max_bytes: 1000mb

These are my pipelines

Iam sending logs to the kafka forward pipeline like this.. To a separate pipeline from main_input pipeline. main_input pipeline is the entry point of logs


output {

     if ("forward_event" not in [tags])
     {
         elasticsearch { 
            hosts => ["localhost:9200"] 
            index => "ScribblerCurrent"
         }
     }



    if ("forward_event" in [tags]) {
        pipeline {
            send_to => forward_Kafka
        }
    }

}