fluentd kafka plugin fails to connect to kafka server using sasl(kerberos) authentication

[Naveenrajp26]

What is a problem?

i am trying to push the data to kafka server which is setup with kerberos authentication and i am facing below error in sending the data. please suggest

fluentd logs:
{"time":"2025-08-03T02:24:59.891530Z","level":"warn","message":"rdkafka: [thrd:sim0142node02.tre.nsn-rdnet.net:9093/bootstrap]: :9093/bootstrap: Disconnected: verify that security.protocol is correctly configured, broker might require SASL authentication (after 340ms in state UP, 4 identical error(s) suppressed)","worker_id":3}
{"time":"2025-08-03T02:25:00.370518Z","level":"warn","message":"rdkafka: [thrd:sim0142node02.tre.nsn-rdnet.net:9093/bootstrap]: :9093/bootstrap: Disconnected: verify that security.protocol is correctly configured, broker might require SASL authentication (after 257ms in state UP, 4 identical error(s) suppressed)","worker_id":3}

version of kafka packages:
bash-5.1$ fluent-gem list | grep kafka
fluent-plugin-kafka (0.19.5, 0.19.2)
rdkafka (0.12.0)
ruby-kafka (1.5.0)
bash-5.1$

fluentd.conf
bash-5.1$ cat /etc/fluent/fluentd.conf
#If you have own configuration for fluentd other than provided by belk/clog then set fluentd_config: custom-value and provide your configuration below. Example-

workers 4

format json
time_format %Y-%m-%dT%H:%M:%S.%6NZ

ignore_same_log_interval 30s

<source>
  @type http
  @log_level error
  @id input_http_ipv4
  port 9000
  bind 0.0.0.0
  <transport tls>
    ca_path <<isroot_cert>>
    cert_path <<fluentd_cert>>
    private_key_path <<fluentd_key_cert>>
  </transport>
</source>

<source>
  @type http
  @log_level error
  @id input_http_ipv6
  port 9000
  bind ::
  <transport tls>
    ca_path <<isroot_cert>>
    cert_path <<fluentd_cert>>
    private_key_path <<fluentd_key_cert>>
  </transport>
</source>

<source>
  @type exec
  tag reload
  command sh /opt/scripts/reload-config-and-crt.sh
  run_interval 5s
  keys none
  <parse>
    @type json
  </parse>
</source>

<match reload>
  @type stdout
  <format>
    @type json
  </format>
</match>

@include /etc/fluent/filter.conf

@include /etc/fluent/rdkafka_no_tls.conf

rdkafka_no_tls.conf

@type copy
copy_mode shallow

@type rdkafka2 @log_level debug

brokers <SERVER FQDN>:9093
topic_key topic
default_topic fmkafkanaveentopic

<buffer date, vnf_name>
  @type file
  @log_level error
  path /data/fluentdlogs/fmkafka
  timekey 1d
  flush_thread_count 4
  chunk_limit_size 500KB
  overflow_action drop_oldest_chunk
  flush_mode interval
  flush_interval 5s
  total_limit_size 2GB
</buffer>

<format>
  @type json
</format>

required_acks 1
request_required_acks 1
delivery_timeout 30000
max_send_retries 10
retry_wait 60s
retry_backoff 1

# TLS settings
#ssl.ca.location /etc/fluent/certs/fm_kafka_ca_cert.pem
#ssl.certificate.location /etc/fluent/certs/fm_kafka_client_cert.pem
#ssl.key.location /etc/fluent/certs/fm_kafka_client_cert_key.pem
# ssl.endpoint.identification.algorithm ""  # disables hostname verification


sasl.mechanisms GSSAPI
sasl.kerberos.keytab /etc/fluent/certs/kafka_client.keytab
sasl.kerberos.principal kafka_client/<SERVER FDQN>@TRE.NSN-RDNET.NET
sasl.kerberos.service.name kafka
security.protocol sasl_plaintext

reconnect_on_error true
reload_on_failure true

bash-5.1$ klist
Ticket cache: FILE:/tmp/krb5cc_1000
Default principal: kafka_client/@TRE.NSN-RDNET.NET

Valid starting Expires Service principal
08/03/25 02:08:31 08/04/25 02:08:31 krbtgt/[email protected]
bash-5.1$

cat config/kafka_server_jaas.conf
// Specifies a unique keytab and principal name for each broker
KafkaServer {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/etc/security/keytabs/kafka.keytab"
principal="kafka/@TRE.NSN-RDNET.NET";
};

Client {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/etc/security/keytabs/kafka_client.keytab"
principal="kafka_client/@TRE.NSN-RDNET.NET";
};

server.properties
listeners=SASL_PLAINTEXT://:9093

advertised.listeners=SASL_PLAINTEXT://:9093
listener.security.protocol.map=SASL_PLAINTEXT:SASL_PLAINTEXT

security.inter.broker.protocol=SASL_PLAINTEXT

sasl.kerberos.service.name=kafka

Specify one of of the SASL mechanisms

sasl.mechanism.inter.broker.protocol=GSSAPI

zookeeper.connect=:2181
zookeeper.connection.timeout.ms=18000
zookeeper.sasl.client=false

Describe the configuration of Fluentd

No response

Describe the logs of Fluentd

No response

Environment

- Fluentd version:
- TD Agent version:
- Fluent Package version:
- Docker image (tag):
- Operating system:
- Kernel version: