out_kafka/out_kafka2/out_kafka_buffered: Provide hash function choices

Signed-off-by: Hiroshi Hatake <hatake@calyptia.com>

Author: cosmo0920

README.md

@@ -202,6 +202,7 @@ If `ruby-kafka` doesn't fit your kafka environment, check `rdkafka2` plugin inst
   headers_from_record (hash) :default => {}
   use_default_for_unknown_topic (bool) :default => false
   discard_kafka_delivery_failed (bool) :default => false (No discard)
+  partitioner_hash_function (enum) (crc32|murmur2) :default => 'crc32'
 
   <format>
   @type (json|ltsv|msgpack|attr:<record name>|<formatter name>) :default => json
@@ -343,6 +344,7 @@ Support of fluentd v0.12 has ended. `kafka_buffered` will be an alias of `kafka2
   exclude_topic_key (bool) :default => false
   exclude_partition_key (bool) :default => false
   get_kafka_client_log (bool) :default => false
+  partitioner_hash_function (enum) (crc32|murmur2) :default => 'crc32'
 
   # See fluentd document for buffer related parameters: https://docs.fluentd.org/v/0.12/buffer
 
@@ -385,6 +387,7 @@ This plugin uses ruby-kafka producer for writing data. For performance and relia
   output_include_time (bool) :default => false
   exclude_topic_key (bool) :default => false
   exclude_partition_key (bool) :default => false
+  partitioner_hash_function (enum) (crc32|murmur2) :default => 'crc32'
 
   # ruby-kafka producer options
   max_send_retries (integer) :default => 1

lib/fluent/plugin/out_kafka.rb

@@ -19,6 +19,8 @@ class Fluent::KafkaOutput < Fluent::Output
  config_param :default_message_key, :string, :default => nil
  config_param :default_partition_key, :string, :default => nil
  config_param :default_partition, :integer, :default => nil
+ config_param :partitioner_hash_function, :enum, list: [:crc32, :murmur2], :default => :crc32,
+ :desc => "Specify kafka patrtitioner hash algorithm"
  config_param :client_id, :string, :default => 'kafka'
  config_param :sasl_over_ssl, :bool, :default => true,
  :desc => <<-DESC
@@ -109,15 +111,18 @@ def refresh_client
  @kafka = Kafka.new(seed_brokers: @seed_brokers, client_id: @client_id, 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_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)
+ ssl_verify_hostname: @ssl_verify_hostname,
+ 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, 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_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)
+ sasl_plain_username: @username, sasl_plain_password: @password, sasl_over_ssl: @sasl_over_ssl, ssl_verify_hostname: @ssl_verify_hostname,
+ partitioner: Kafka::Partitioner.new(hash_function: @partitioner_hash_function))
  else
  @kafka = Kafka.new(seed_brokers: @seed_brokers, client_id: @client_id, 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_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)
+ sasl_gssapi_principal: @principal, sasl_gssapi_keytab: @keytab, sasl_over_ssl: @sasl_over_ssl, ssl_verify_hostname: @ssl_verify_hostname,
+ partitioner: Kafka::Partitioner.new(hash_function: @partitioner_hash_function))
  end
  log.info "initialized kafka producer: #{@client_id}"
  else

lib/fluent/plugin/out_kafka2.rb

@@ -24,6 +24,8 @@ class Fluent::Kafka2Output < Output
  config_param :partition_key_key, :string, :default => 'partition_key', :desc => "Field for kafka partition key"
  config_param :default_partition_key, :string, :default => nil
  config_param :partition_key, :string, :default => 'partition', :desc => "Field for kafka partition"
+ config_param :partitioner_hash_function, :enum, list: [:crc32, :murmur2], :default => :crc32,
+ :desc => "Specify kafka patrtitioner hash algorithm"
  config_param :default_partition, :integer, :default => nil
  config_param :use_default_for_unknown_topic, :bool, :default => false, :desc => "If true, default_topic is used when topic not found"
  config_param :client_id, :string, :default => 'fluentd'
@@ -99,17 +101,20 @@ def refresh_client(raise_error = true)
  @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_scram_username: @username, sasl_scram_password: @password,
- sasl_scram_mechanism: @scram_mechanism, sasl_over_ssl: @sasl_over_ssl, ssl_verify_hostname: @ssl_verify_hostname)
+ sasl_scram_mechanism: @scram_mechanism, sasl_over_ssl: @sasl_over_ssl, ssl_verify_hostname: @ssl_verify_hostname,
+ 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_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)
+ ssl_verify_hostname: @ssl_verify_hostname,
+ 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_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)
+ ssl_verify_hostname: @ssl_verify_hostname,
+ partitioner: Kafka::Partitioner.new(hash_function: @partitioner_hash_function))
  end
  log.info "initialized kafka producer: #{@client_id}"
  rescue Exception => e

lib/fluent/plugin/out_kafka_buffered.rb

@@ -26,6 +26,8 @@ class Fluent::KafkaOutputBuffered < Fluent::BufferedOutput
  config_param :default_partition_key, :string, :default => nil
  config_param :partition_key, :string, :default => 'partition', :desc => "Field for kafka partition"
  config_param :default_partition, :integer, :default => nil
+ config_param :partitioner_hash_function, :enum, list: [:crc32, :murmur2], :default => :crc32,
+ :desc => "Specify kafka patrtitioner hash algorithm"
  config_param :client_id, :string, :default => 'kafka'
  config_param :idempotent, :bool, :default => false, :desc => 'Enable idempotent producer'
  config_param :sasl_over_ssl, :bool, :default => true,
@@ -133,15 +135,18 @@ def refresh_client(raise_error = true)
  @kafka = Kafka.new(seed_brokers: @seed_brokers, client_id: @client_id, logger: logger, 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_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)
+ ssl_verify_hostname: @ssl_verify_hostname,
+ 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, 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_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)
+ sasl_plain_username: @username, sasl_plain_password: @password, sasl_over_ssl: @sasl_over_ssl, ssl_verify_hostname: @ssl_verify_hostname,
+ partitioner: Kafka::Partitioner.new(hash_function: @partitioner_hash_function))
  else
  @kafka = Kafka.new(seed_brokers: @seed_brokers, client_id: @client_id, logger: logger, 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_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)
+ sasl_gssapi_principal: @principal, sasl_gssapi_keytab: @keytab, sasl_over_ssl: @sasl_over_ssl, ssl_verify_hostname: @ssl_verify_hostname,
+ partitioner: Kafka::Partitioner.new(hash_function: @partitioner_hash_function))
  end
  log.info "initialized kafka producer: #{@client_id}"
  else