Logstash Drops Buffered Data on Kafka Authentication Failure

[robbavey]

(Transfered from elastic/logstash#18560)

Logstash information:

Logstash version - 8.19.3
Logstash installation source - Extracted from zip archive
Logstash being run as service

Plugins installed:

logstash-codec-avro (3.4.1)
logstash-codec-cef (6.2.8)
logstash-codec-cefpartial (0.3.4)
logstash-codec-collectd (3.1.0)
logstash-codec-dots (3.0.6)
logstash-codec-edn (3.1.0)
logstash-codec-edn_lines (3.1.0)
logstash-codec-es_bulk (3.1.0)
logstash-codec-fluent (3.4.3)
logstash-codec-graphite (3.0.6)
logstash-codec-json (3.1.1)
logstash-codec-json_lines (3.2.2)
logstash-codec-line (3.1.1)
logstash-codec-msgpack (3.1.0)
logstash-codec-multiline (3.1.2)
logstash-codec-netflow (4.3.2)
logstash-codec-plain (3.1.0)
logstash-codec-rubydebug (3.1.0)
logstash-codec-syslogx (0.5.2)
logstash-filter-aggregate (2.10.0)
logstash-filter-anonymize (3.0.7)
logstash-filter-cidr (3.1.3)
logstash-filter-clone (4.2.0)
logstash-filter-csv (3.1.1)
logstash-filter-date (3.1.15)
logstash-filter-dateruby (0.2.3)
logstash-filter-de_dot (1.1.0)
logstash-filter-dissect (1.2.5)
logstash-filter-dns (3.2.0)
logstash-filter-drop (3.0.5)
logstash-filter-elasticsearch (3.18.0)
logstash-filter-fingerprint (3.4.4)
logstash-filter-geoip (7.3.1)
logstash-filter-grok (4.4.3)
logstash-filter-http (1.6.0)
logstash-filter-json (3.2.1)
logstash-filter-kv (4.7.0)
logstash-filter-memcached (1.2.0)
logstash-filter-metrics (4.0.7)
logstash-filter-mutate (3.5.8)
logstash-filter-prune (3.0.4)
logstash-filter-ruby (3.1.8)
logstash-filter-sleep (3.0.7)
logstash-filter-snmptrap (0.9.11)
logstash-filter-split (3.1.8)
logstash-filter-syslog_pri (3.2.1)
logstash-filter-throttle (4.0.4)
logstash-filter-translate (3.4.3)
logstash-filter-truncate (1.0.6)
logstash-filter-urldecode (3.0.6)
logstash-filter-useragent (3.3.5)
logstash-filter-uuid (3.0.5)
logstash-filter-xml (4.3.2)
logstash-input-azure_event_hubs (1.5.2)
logstash-input-beats (6.9.3)
└── logstash-input-elastic_agent (alias)
logstash-input-couchdb_changes (3.1.6)
logstash-input-dead_letter_queue (2.0.1)
logstash-input-elastic_serverless_forwarder (1.0.0)
logstash-input-elasticsearch (4.23.0)
logstash-input-exec (3.6.0)
logstash-input-file (4.4.6)
logstash-input-ganglia (3.1.4)
logstash-input-gelf (3.3.2)
logstash-input-generator (3.1.0)
logstash-input-graphite (3.0.6)
logstash-input-heartbeat (3.1.1)
logstash-input-http (3.10.2)
logstash-input-http_poller (5.6.1)
logstash-input-jms (3.3.0)
logstash-input-pipe (3.1.0)
logstash-input-redis (3.7.1)
logstash-input-stdin (3.4.0)
logstash-input-syslog (3.7.1)
logstash-input-tcp (6.4.6)
logstash-input-twitter (4.1.1)
logstash-input-udp (3.5.0)
logstash-input-unix (3.1.2)
logstash-integration-aws (7.2.1)
├── logstash-codec-cloudfront
├── logstash-codec-cloudtrail
├── logstash-input-cloudwatch
├── logstash-input-s3
├── logstash-input-sqs
├── logstash-output-cloudwatch
├── logstash-output-s3
├── logstash-output-sns
└── logstash-output-sqs
logstash-integration-elastic_enterprise_search (3.0.1)
├── logstash-output-elastic_app_search
└── logstash-output-elastic_workplace_search
logstash-integration-jdbc (5.6.0)
├── logstash-input-jdbc
├── logstash-filter-jdbc_streaming
└── logstash-filter-jdbc_static
logstash-integration-kafka (11.6.3)
├── logstash-input-kafka
└── logstash-output-kafka
logstash-integration-logstash (1.0.4)
├── logstash-input-logstash
└── logstash-output-logstash
logstash-integration-rabbitmq (7.4.0)
├── logstash-input-rabbitmq
└── logstash-output-rabbitmq
logstash-integration-snmp (4.0.7)
├── logstash-input-snmp
└── logstash-input-snmptrap
logstash-output-csv (3.0.10)
logstash-output-elasticsearch (11.22.13)
logstash-output-email (4.1.3)
logstash-output-file (4.3.0)
logstash-output-graphite (3.1.6)
logstash-output-http (5.7.1)
logstash-output-lumberjack (3.1.9)
logstash-output-nagios (3.0.6)
logstash-output-null (3.0.5)
logstash-output-pipe (3.0.6)
logstash-output-redis (5.2.0)
logstash-output-splunk (1.2.2)
logstash-output-stdout (3.1.4)
logstash-output-syslog (4.2.3)
logstash-output-tcp (6.2.2)
logstash-output-udp (3.2.0)
logstash-output-webhdfs (3.1.0)
logstash-patterns-core (4.3.4)

JVM:

openjdk version "21.0.7" 2025-04-15 LTS

OS version:

Windows Server 2025

Description of the problem including expected versus actual behavior:

Logstash is used to forward events to a Kafka cluster secured with SASL/PLAIN authentication. When valid Kafka credentials are configured, Logstash successfully delivers all events to Kafka. For failure-handling testing, incorrect Kafka credentials are configured in the Logstash Kafka output. During this period, Logstash continues to receive events and temporarily buffers them while attempting to authenticate with Kafka.

After repeated authentication failures, Logstash begins dropping events that are already present in the buffer. The Logstash logs show repeated producer send failed, dropping record messages caused by SaslAuthenticationException.

Expected behavior is that, when Kafka authentication fails, Logstash should retain buffered events and continue retrying delivery without dropping data, and once valid credentials are restored, all buffered events should be delivered to Kafka.

Steps to reproduce:

Forward PipeLine

input {
	generator {
		message => '{"test":"logstash-kafka-sasl-repro"}'
		count => 1000
	}
}

output {

	kafka {
	
		codec => json
		topic_id => "string"
		bootstrap_servers => "kafka.example:7880"
	
		client_id => "Ragul-kumar"
		security_protocol => "SASL_PLAINTEXT"
		
		#SSL Truststore
		
		
		
		# SASL configuration (enabled only when security_protocol is set to SASL_SSL)
		
						sasl_mechanism => "PLAIN"
						 sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='kafkaUser' password='Abcd@12';"           # wrong password
	
		
	}
}

Logs

[2026-01-06T00:00:32,018][INFO ][org.apache.kafka.clients.NetworkClient][scribbler_forward_log_cockpit] [Producer clientId=scribbler-WIN-AH9POJL038C] Node -1 disconnected.
[2026-01-06T00:00:32,018][ERROR][org.apache.kafka.clients.NetworkClient][scribbler_forward_log_cockpit] [Producer clientId=scribbler-WIN-AH9POJL038C] Connection to node -1 (kafka.example/x.x.x.x:7880) failed authentication due to: Authentication failed: Invalid username or password
[2026-01-06T00:00:32,018][WARN ][org.apache.kafka.clients.NetworkClient][scribbler_forward_log_cockpit] [Producer clientId=scribbler-WIN-AH9POJL038C] Bootstrap broker kafka.example:7880 (id: -1 rack: null) disconnected
[2026-01-06T00:00:32,433][INFO ][org.apache.kafka.common.network.Selector][scribbler_forward_log_cockpit] [Producer clientId=scribbler-WIN-AH9POJL038C] Failed authentication with kafka.example/x.x.x.x (channelId=-1) (Authentication failed: Invalid username or password)
[2026-01-06T00:00:32,434][INFO ][org.apache.kafka.clients.NetworkClient][scribbler_forward_log_cockpit] [Producer clientId=scribbler-WIN-AH9POJL038C] Node -1 disconnected.
[2026-01-06T00:00:32,434][ERROR][org.apache.kafka.clients.NetworkClient][scribbler_forward_log_cockpit] [Producer clientId=scribbler-WIN-AH9POJL038C] Connection to node -1 (kafka.example/x.x.x.x:7880) failed authentication due to: Authentication failed: Invalid username or password
[2026-01-06T00:00:32,434][WARN ][org.apache.kafka.clients.NetworkClient][scribbler_forward_log_cockpit] [Producer clientId=scribbler-WIN-AH9POJL038C] Bootstrap broker kafka.example:7880 (id: -1 rack: null) disconnected
[2026-01-06T00:00:32,441][WARN ][logstash.outputs.kafka   ][scribbler_forward_log_cockpit][22cf8ad841f5f9ea736a75674a02ca338d70ea00b74fc20b62cf27a69c28dd2b] producer send failed, dropping record {:exception=>Java::OrgApacheKafkaCommonErrors::SaslAuthenticationException, :message=>"Authentication failed: Invalid username or password", :record_value=>"{\"event\":{\"created\":\"2026-01-06T08:00:28.340239200Z\",\"kind\":\"event\",\"severity\":\"5\"},\"@version\":\"1\",\"@timestamp\":\"2026-01-06T08:00:28.340239200Z\",\"labels\":{\"event_source_type\":\"syslog\",\"event_source_format\":\"Other\"},\"host\":{\"name\":\"10.10.1.238\",\"ip\":\"10.10.1.238\"},\"observer\":{\"type\":\"Logger\",\"vendor\":\"SyskeyOT\",\"hostname\":\"WIN-AH9POJL038C\",\"geo\":{\"timezone\":\"Asia/Calcutta\"},\"version\":\"1.15.21\",\"product\":\"Scribbler Log Manager\"},\"log\":{\"syslog\":{\"facility\":{\"name\":\"user-level\",\"code\":\"1\"},\"severity\":{\"name\":\"notice\",\"code\":\"5\"},\"priority\":\"13\"}},\"message\":\"Sigma Ragul\"}"}
[2026-01-06T00:00:32,443][WARN ][logstash.outputs.kafka   ][scribbler_forward_log_cockpit][22cf8ad841f5f9ea736a75674a02ca338d70ea00b74fc20b62cf27a69c28dd2b] producer send failed, dropping record {:exception=>Java::OrgApacheKafkaCommonErrors::SaslAuthenticationException, :message=>"Authentication failed: Invalid username or password", :record_value=>"{\"event\":{\"created\":\"2026-01-06T08:00:28.413106300Z\",\"kind\":\"event\",\"severity\":\"5\"},\"@version\":\"1\",\"@timestamp\":\"2026-01-06T08:00:28.413106300Z\",\"labels\":{\"event_source_type\":\"syslog\",\"event_source_format\":\"Other\"},\"host\":{\"name\":\"10.10.1.238\",\"ip\":\"10.10.1.238\"},\"observer\":{\"type\":\"Logger\",\"vendor\":\"SyskeyOT\",\"hostname\":\"WIN-AH9POJL038C\",\"geo\":{\"timezone\":\"Asia/Calcutta\"},\"version\":\"1.15.21\",\"product\":\"Scribbler Log Manager\"},\"log\":{\"syslog\":{\"facility\":{\"name\":\"user-level\",\"code\":\"1\"},\"severity\":{\"name\":\"notice\",\"code\":\"5\"},\"priority\":\"13\"}},\"message\":\"Sigma Ragul\"}"}
[2026-01-06T00:00:32,446][WARN ][logstash.outputs.kafka   ][scribbler_forward_log_cockpit][22cf8ad841f5f9ea736a75674a02ca338d70ea00b74fc20b62cf27a69c28dd2b] producer send failed, dropping record {:exception=>Java::OrgApacheKafkaCommonErrors::SaslAuthenticationException, :message=>"Authentication failed: Invalid username or password", :record_value=>"{\"event\":{\"created\":\"2026-01-06T08:00:28.413106300Z\",\"kind\":\"event\",\"severity\":\"5\"},\"@version\":\"1\",\"@timestamp\":\"2026-01-06T08:00:28.413106300Z\",\"labels\":{\"event_source_type\":\"syslog\",\"event_source_format\":\"Other\"},\"host\":{\"name\":\"10.10.1.238\",\"ip\":\"10.10.1.238\"},\"observer\":{\"type\":\"Logger\",\"vendor\":\"SyskeyOT\",\"hostname\":\"WIN-AH9POJL038C\",\"geo\":{\"timezone\":\"Asia/Calcutta\"},\"version\":\"1.15.21\",\"product\":\"Scribbler Log Manager\"},\"log\":{\"syslog\":{\"facility\":{\"name\":\"user-level\",\"code\":\"1\"},\"severity\":{\"name\":\"notice\",\"code\":\"5\"},\"priority\":\"13\"}},\"message\":\"Sigma Ragul\"}"}
[2026-01-06T00:00:32,446][WARN ][logstash.outputs.kafka   ][scribbler_forward_log_cockpit][22cf8ad841f5f9ea736a75674a02ca338d70ea00b74fc20b62cf27a69c28dd2b] producer send failed, dropping record {:exception=>Java::OrgApacheKafkaCommonErrors::SaslAuthenticationException, :message=>"Authentication failed: Invalid username or password", :record_value=>"{\"event\":{\"created\":\"2026-01-06T08:00:28.413106300Z\",\"kind\":\"event\",\"severity\":\"5\"},\"@version\":\"1\",\"@timestamp\":\"2026-01-06T08:00:28.413106300Z\",\"labels\":{\"event_source_type\":\"syslog\",\"event_source_format\":\"Other\"},\"host\":{\"name\":\"10.10.1.238\",\"ip\":\"10.10.1.238\"},\"observer\":{\"type\":\"Logger\",\"vendor\":\"SyskeyOT\",\"hostname\":\"WIN-AH9POJL038C\",\"geo\":{\"timezone\":\"Asia/Calcutta\"},\"version\":\"1.15.21\",\"product\":\"Scribbler Log Manager\"},\"log\":{\"syslog\":{\"facility\":{\"name\":\"user-level\",\"code\":\"1\"},\"severity\":{\"name\":\"notice\",\"code\":\"5\"},\"priority\":\"13\"}},\"message\":\"Sigma Ragul\"}"}
[2026-01-06T00:00:32,447][WARN ][logstash.outputs.kafka   ][scribbler_forward_log_cockpit][22cf8ad841f5f9ea736a75674a02ca338d70ea00b74fc20b62cf27a69c28dd2b] producer send failed, dropping record {:exception=>Java::OrgApacheKafkaCommonErrors::SaslAuthenticationException, :message=>"Authentication failed: Invalid username or password", :record_value=>"{\"event\":{\"created\":\"2026-01-06T08:00:28.415896900Z\",\"kind\":\"event\",\"severity\":\"5\"},\"@version\":\"1\",\"@timestamp\":\"2026-01-06T08:00:28.415896900Z\",\"labels\":{\"event_source_type\":\"syslog\",\"event_source_format\":\"Other\"},\"host\":{\"name\":\"10.10.1.238\",\"ip\":\"10.10.1.238\"},\"observer\":{\"type\":\"Logger\",\"vendor\":\"SyskeyOT\",\"hostname\":\"WIN-AH9POJL038C\",\"geo\":{\"timezone\":\"Asia/Calcutta\"},\"version\":\"1.15.21\",\"product\":\"Scribbler Log Manager\"},\"log\":{\"syslog\":{\"facility\":{\"name\":\"user-level\",\"code\":\"1\"},\"severity\":{\"name\":\"notice\",\"code\":\"5\"},\"priority\":\"13\"}},\"message\":\"Sigma Ragul\"}"}
[2026-01-06T00:00:32,448][WARN ][logstash.outputs.kafka   ][scribbler_forward_log_cockpit][22cf8ad841f5f9ea736a75674a02ca338d70ea00b74fc20b62cf27a69c28dd2b] producer send failed, dropping record