out_kafka2: Unify duplicate kafka params

Signed-off-by: Takuro Ashie <[email protected]>

Author: ashie

lib/fluent/plugin/out_kafka2.rb

@@ -116,84 +116,45 @@ def refresh_client(raise_error = true)
         logger = @get_kafka_client_log ? log : nil
         use_long_lived_aws_credentials = @sasl_aws_msk_iam_access_key_id != nil && @sasl_aws_msk_iam_secret_key_id != nil
         if @scram_mechanism != nil && @username != nil && @password != nil
-          @kafka = Kafka.new(
-            seed_brokers: @seed_brokers,
-            client_id: @client_id,
-            logger: logger,
-            connect_timeout: @connect_timeout,
-            socket_timeout: @socket_timeout,
-            ssl_ca_cert_file_path: @ssl_ca_cert,
-            ssl_client_cert: read_ssl_file(@ssl_client_cert),
-            ssl_client_cert_key: read_ssl_file(@ssl_client_cert_key),
-            ssl_client_cert_key_password: @ssl_client_cert_key_password,
-            ssl_client_cert_chain: read_ssl_file(@ssl_client_cert_chain),
-            ssl_ca_certs_from_system: @ssl_ca_certs_from_system,
+          sasl_params = {
             sasl_scram_username: @username,
             sasl_scram_password: @password,
             sasl_scram_mechanism: @scram_mechanism,
-            sasl_over_ssl: @sasl_over_ssl,
-            ssl_verify_hostname: @ssl_verify_hostname,
-            resolve_seed_brokers: @resolve_seed_brokers,
-            partitioner: Kafka::Partitioner.new(hash_function: @partitioner_hash_function))
+          }
         elsif @username != nil && @password != nil
-          @kafka = Kafka.new(
-            seed_brokers: @seed_brokers,
-            client_id: @client_id,
-            logger: logger,
-            connect_timeout: @connect_timeout,
-            socket_timeout: @socket_timeout,
-            ssl_ca_cert_file_path: @ssl_ca_cert,
-            ssl_client_cert: read_ssl_file(@ssl_client_cert),
-            ssl_client_cert_key: read_ssl_file(@ssl_client_cert_key),
-            ssl_client_cert_key_password: @ssl_client_cert_key_password,
-            ssl_client_cert_chain: read_ssl_file(@ssl_client_cert_chain),
-            ssl_ca_certs_from_system: @ssl_ca_certs_from_system,
+          sasl_params = {
             sasl_plain_username: @username,
             sasl_plain_password: @password,
-            sasl_over_ssl: @sasl_over_ssl,
-            ssl_verify_hostname: @ssl_verify_hostname,
-            resolve_seed_brokers: @resolve_seed_brokers,
-            partitioner: Kafka::Partitioner.new(hash_function: @partitioner_hash_function))
+          }
         elsif use_long_lived_aws_credentials
-          @kafka = Kafka.new(
-            seed_brokers: @seed_brokers,
-            client_id: @client_id,
-            logger: logger,
-            connect_timeout: @connect_timeout,
-            socket_timeout: @socket_timeout,
-            ssl_ca_cert_file_path: @ssl_ca_cert,
-            ssl_client_cert: read_ssl_file(@ssl_client_cert),
-            ssl_client_cert_key: read_ssl_file(@ssl_client_cert_key),
-            ssl_client_cert_chain: read_ssl_file(@ssl_client_cert_chain),
-            ssl_ca_certs_from_system: @ssl_ca_certs_from_system,
-            sasl_over_ssl: @sasl_over_ssl,
-            ssl_verify_hostname: @ssl_verify_hostname,
-            resolve_seed_brokers: @resolve_seed_brokers,
+          sasl_params = {
             sasl_aws_msk_iam_access_key_id: @sasl_aws_msk_iam_access_key_id,
             sasl_aws_msk_iam_secret_key_id: @sasl_aws_msk_iam_secret_key_id,
             sasl_aws_msk_iam_aws_region: @sasl_aws_msk_iam_aws_region,
-            partitioner: Kafka::Partitioner.new(hash_function: @partitioner_hash_function)
-          )
+          }
         else
-          @kafka = Kafka.new(
-            seed_brokers: @seed_brokers,
-            client_id: @client_id,
-            logger: logger,
-            connect_timeout: @connect_timeout,
-            socket_timeout: @socket_timeout,
-            ssl_ca_cert_file_path: @ssl_ca_cert,
-            ssl_client_cert: read_ssl_file(@ssl_client_cert),
-            ssl_client_cert_key: read_ssl_file(@ssl_client_cert_key),
-            ssl_client_cert_key_password: @ssl_client_cert_key_password,
-            ssl_client_cert_chain: read_ssl_file(@ssl_client_cert_chain),
-            ssl_ca_certs_from_system: @ssl_ca_certs_from_system,
+          sasl_params = {
             sasl_gssapi_principal: @principal,
             sasl_gssapi_keytab: @keytab,
-            sasl_over_ssl: @sasl_over_ssl,
-            ssl_verify_hostname: @ssl_verify_hostname,
-            resolve_seed_brokers: @resolve_seed_brokers,
-            partitioner: Kafka::Partitioner.new(hash_function: @partitioner_hash_function))
+          }
         end
+        @kafka = Kafka.new(
+          seed_brokers: @seed_brokers,
+          client_id: @client_id,
+          logger: logger,
+          connect_timeout: @connect_timeout,
+          socket_timeout: @socket_timeout,
+          ssl_ca_cert_file_path: @ssl_ca_cert,
+          ssl_client_cert: read_ssl_file(@ssl_client_cert),
+          ssl_client_cert_key: read_ssl_file(@ssl_client_cert_key),
+          ssl_client_cert_key_password: @ssl_client_cert_key_password,
+          ssl_client_cert_chain: read_ssl_file(@ssl_client_cert_chain),
+          ssl_ca_certs_from_system: @ssl_ca_certs_from_system,
+          ssl_verify_hostname: @ssl_verify_hostname,
+          resolve_seed_brokers: @resolve_seed_brokers,
+          partitioner: Kafka::Partitioner.new(hash_function: @partitioner_hash_function),
+          sasl_over_ssl: @sasl_over_ssl,
+          **sasl_params)
         log.info "initialized kafka producer: #{@client_id}"
       rescue Exception => e
         if raise_error # During startup, error should be reported to engine and stop its phase for safety.