*Description of problem:

Kafka Authentication failed: Invalid username or password . The fluentd.conf are generated by CLO. username/password are enabled by ruby expression .

Once set the username/password using hardcode. fluentd can talks with kafka

Ruby expression:
username "#

"
password "#

"

Version:
quay.io/logging/origin-cluster-logging-operator:latest 202101090039.11723
clo image build using latest master branch.

*How to reproducible:
Always

*Step to Reproduce
1) git clone git@gitlab.cee.redhat.com:aosqe/aosqe-tools.git
2) cd osqe-tools/logging/log_template/kafka/kafka-2.4.1
3) Deploy clusterlogging operator
4) Create certificates

01_create-pki-cluster-client.sh

5) Deploy Kafka and enable sasl ssl

10_deploy-kafka-plaintext-sasl_ssl.sh

6) Forward logs to kafka using username and passworkd

20_create-clf-kafka-sasl_ssl.sh

7) Check all fluentd pods logs, check consumer pods logs, check the kafka logs lods

Actual Result:
fluentd logs
2021-05-16 15:28:47 +0000 [warn]: failed to flush the buffer. retry_time=29 next_retry_seconds=2021-05-16 15:29:45 +0000 chunk="5c273b7a8dcbf1b468fb559be3c6939c" error_class=Kafka::ConnectionError error="Could not connect to any of the seed brokers:\n- kafka://kafka.openshift-logging.svc.cluster.local:9093: SASL PLAIN authentication failed: end of file reached"
2021-05-16 15:28:47 +0000 [warn]: suppressed same stacktrace
2021-05-16 15:28:47 +0000 [warn]: failed to flush the buffer. retry_time=30 next_retry_seconds=2021-05-16 15:29:40 +0000 chunk="5c273b7b81f0787933077eab3c8ca72c" error_class=Kafka::ConnectionError error="Could not connect to any of the seed brokers:\n- kafka://kafka.openshift-logging.svc.cluster.local:9093: SASL PLAIN authentication failed: end of file reached"
2021-05-16 15:28:47 +0000 [warn]: suppressed same stacktrace
2021-05-16 15:29:41 +0000 [warn]: Send exception occurred: Could not connect to any of the seed brokers:

• kafka://kafka.openshift-logging.svc.cluster.local:9093: SASL PLAIN authentication failed: end of file reached
  2021-05-16 15:29:41 +0000 [warn]: Exception Backtrace : /usr/local/share/gems/gems/ruby-kafka-1.1.0/lib/kafka/cluster.rb:448:in `fetch_cluster_info'
  /usr/local/share/gems/gems/ruby-kafka-1.1.0/lib/kafka/cluster.rb:402:in `cluster_info'
  /usr/local/share/gems/gems/ruby-kafka-1.1.0/lib/kafka/cluster.rb:102:in `refresh_metadata!'
  /usr/local/share/gems/gems/ruby-kafka-1.1.0/lib/kafka/cluster.rb:56:in `add_target_topics'
  /usr/local/share/gems/gems/fluent-plugin-kafka-0.13.1/lib/fluent/plugin/kafka_producer_ext.rb:91:in `initialize'
  /usr/local/share/gems/gems/fluent-plugin-kafka-0.13.1/lib/fluent/plugin/kafka_producer_ext.rb:60:in `new'
  /usr/local/share/gems/gems/fluent-plugin-kafka-0.13.1/lib/fluent/plugin/kafka_producer_ext.rb:60:in `topic_producer'
  /usr/local/share/gems/gems/fluent-plugin-kafka-0.13.1/lib/fluent/plugin/out_kafka2.rb:232:in `write'
  /usr/local/share/gems/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:1125:in `try_flush'
  /usr/local/share/gems/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:1431:in `flush_thread_run'
  /usr/local/share/gems/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:461:in `block (2 levels) in start'
  /usr/local/share/gems/gems/fluentd-1.7.4/lib/fluent/plugin_helper/thread.rb:78:in `block in thread_create'
  2021-05-16 15:29:41 +0000 [warn]: failed to flush the buffer. retry_time=31 next_retry_seconds=2021-05-16 15:30:33 +0000 chunk="5c273b7b81f0787933077eab3c8ca72c" error_class=Kafka::ConnectionError error="Could not connect to any of the seed brokers:\n- kafka://kafka.openshift-logging.svc.cluster.local:9093: SASL PLAIN authentication failed: end of file reached"

kafka logs

[2021-05-16 15:28:47,272] INFO [SocketServer brokerId=0] Failed authentication with 10-128-3-184.fluentd.openshift-logging.svc.cluster.local/10.128.3.184 (Authentication failed: Invalid username or password) (org.apache.kafka.common.network.Selector)
[2021-05-16 15:28:47,272] INFO [SocketServer brokerId=0] Failed authentication with 10-128-3-184.fluentd.openshift-logging.svc.cluster.local/10.128.3.184 (Authentication failed: Invalid username or password) (org.apache.kafka.common.network.Selector)
[2021-05-16 15:29:41,036] INFO [SocketServer brokerId=0] Failed authentication with 10-128-3-184.fluentd.openshift-logging.svc.cluster.local/10.128.3.184 (Authentication failed: Invalid username or password) (org.apache.kafka.common.network.Selector)
[2021-05-16 15:30:33,995] INFO [SocketServer brokerId=0] Failed authentication with 10-128-3-184.fluentd.openshift-logging.svc.cluster.local/10.128.3.184 (Authentication failed: Invalid username or password) (org.apache.kafka.common.network.Selector)

Expected Result
The logs can be forward to Kafka via SASL_SSL