Kafka Output Configuration (unified for both sources)
output.kafka:
hosts: ["xxxxxxx:19094"] # Updated to use the SSL-enabled broker
topic: 'npc-raw-msg'
key: '%{[UUID]}'
codec.format:
string: '%{[message]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
close_inactive: 10m
Enhanced Security Configuration
sasl.mechanism: SCRAM-SHA-512
sasl.username: "admin"
sasl.password: "admin-psswd"
security.protocol: SASL_PLAINTEXT
#ssl.verification_mode: full
#ssl.certificate_authorities: ["/etc/filebeat/certs/ca-cert.pem"]
#ssl.verification_mode: none
timeout: 60s
Retry configuration for better reliability
max_retries: 3
backoff.init: 1s
backoff.max: 60s
"log.level":"error","@timestamp":"2025-10-28T06:20:15.900-0400","log.logger":"kafka","log.origin":{"function":"github.com/elastic/beats/v7/libbeat/outputs/kafka.(*client).errorWorker","file.name":"kafka/client.go","file.line":338},"message":"Kafka (topic=npc-raw-msg): kafka: couldn't fetch broker metadata (check that your client and broker are using the same encryption and authentication settings)","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"error","@timestamp":"2025-10-28T06:20:15.900-0400","log.logger":"kafka","log.origin":{"function":"github.com/elastic/beats/v7/libbeat/outputs/kafka.(*client).errorWorker","file.name":"kafka/client.go","file.line":338},"message":"Kafka (topic=npc-raw-msg): kafka: couldn't fetch broker metadata (check that your client and broker are using the same encryption and authentication settings)","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"error","@timestamp":"2025-10-28T06:20:15.900-0400","log.logger":"kafka","log.origin":{"function":"github.com/elastic/beats/v7/libbeat/outputs/kafka.(*client).errorWorker","file.name":"kafka/client.go","file.line":338},"message":"Kafka (topic=npc-raw-msg): kafka: couldn't fetch broker metadata (check that your client and broker are using the same encryption and authentication settings)","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"error","@timestamp":"2025-10-28T06:20:15.900-0400","log.logger":"kafka","log.origin":{"function":"github.com/elastic/beats/v7/libbeat/outputs/kafka.(*client).errorWorker","file.name":"kafka/client.go","file.line":338},"message":"Kafka (topic=npc-raw-msg): kafka: couldn't fetch broker metadata (check that your client and broker are using the same encryption and authentication settings)","service.name":"filebeat","ecs.version":"1.6.0"}