Merge pull request #482 from fluent/unify-duplicate-kafka-params

Unify duplicate kafka params

Author: ashie

lib/fluent/plugin/out_kafka2.rb

@@ -116,44 +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_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))
+          sasl_params = {
+            sasl_scram_username: @username,
+            sasl_scram_password: @password,
+            sasl_scram_mechanism: @scram_mechanism,
+          }
         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_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))
+          sasl_params = {
+            sasl_plain_username: @username,
+            sasl_plain_password: @password,
+          }
         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_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))
+          sasl_params = {
+            sasl_gssapi_principal: @principal,
+            sasl_gssapi_keytab: @keytab,
+          }
         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.