<commit_before># basically we are porting this https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example

require "java"

require "jruby-kafka/namespace"
require "jruby-kafka/error"

class Kafka::Producer
  java_import 'kafka.producer.ProducerConfig'
  java_import 'kafka.producer.KeyedMessage'
  KafkaProducer = Java::kafka.javaapi.producer.Producer
  java_import 'kafka.message.NoCompressionCodec'
  java_import 'kafka.message.GZIPCompressionCodec'
  java_import 'kafka.message.SnappyCompressionCodec'

  VALIDATIONS = {
    'request.required.acks' => %w[ 0 1 -1 ],
    'required.codecs' => [NoCompressionCodec.name, GZIPCompressionCodec.name, SnappyCompressionCodec.name],
    'producer.type' => %w[ sync async ]
  }

  REQUIRED = %w[
    metadata.broker.list
  ]

  # List of all available options extracted from http://kafka.apache.org/documentation.html#producerconfigs Apr. 27, 2014
  # If new options are added, they should just work. Please add them to the list so that we can get handy warnings.
  KNOWN = %w[
    acks                   max.request.size              receive.buffer.bytes
    batch.num.messages     message.send.max.retries      reconnect.backoff.ms
    batch.size             metadata.broker.list          request.required.acks
    block.on.buffer.full   metadata.fetch.timeout.ms     request.timeout.ms
    bootstrap.servers      metadata.max.age.ms           retries
    buffer.memory          metric.reporters              retry.backoff.ms
    client.id              metrics.num.samples           retry.backoff.ms
    client.id              metrics.sample.window.ms      send.buffer.bytes
    compressed.topics      partitioner.class             send.buffer.bytes
    compression.codec      producer.type                 serializer.class
    compression.type       queue.buffering.max.messages  timeout.ms
    key.serializer.class   queue.buffering.max.ms        topic.metadata.refresh.interval.ms
    linger.ms              queue.enqueue.timeout.ms
  ]

  attr_reader :producer, :send_method, :options

  # Create a Kafka Producer
  #
  # options:
  # metadata_broker_list: ["localhost:9092"] - REQUIRED: a seed list of kafka brokers
  def initialize(opts = {})
    @options = opts.reduce({}) do |opts, (k, v)|
      opts[k.to_s.gsub(/_/, '.')] = v
      opts
    end
    if options['broker.list']
      options['metadata.broker.list'] = options.delete 'broker.list'
    end
    if options['compressed.topics'].to_s == 'none'
      options.delete 'compressed.topics'<commit_msg>make producer options only relevant for Kafka 0.8.x<commit_after># basically we are porting this https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example

require "java"

require "jruby-kafka/namespace"
require "jruby-kafka/error"

class Kafka::Producer
  java_import 'kafka.producer.ProducerConfig'
  java_import 'kafka.producer.KeyedMessage'
  KafkaProducer = Java::kafka.javaapi.producer.Producer
  java_import 'kafka.message.NoCompressionCodec'
  java_import 'kafka.message.GZIPCompressionCodec'
  java_import 'kafka.message.SnappyCompressionCodec'

  VALIDATIONS = {
    'request.required.acks' => %w[ 0 1 -1 ],
    'required.codecs' => [NoCompressionCodec.name, GZIPCompressionCodec.name, SnappyCompressionCodec.name],
    'producer.type' => %w[ sync async ]
  }

  REQUIRED = %w[
    metadata.broker.list
  ]

  # List of all available options extracted from http://kafka.apache.org/documentation.html#producerconfigs Apr. 27, 2014
  # If new options are added, they should just work. Please add them to the list so that we can get handy warnings.
  KNOWN = %w[
    metadata.broker.list      request.required.acks         request.timeout.ms
    producer.type             serializer.class              key.serializer.class
    partitioner.class         compression.codec             compressed.topics
    message.send.max.retries  retry.backoff.ms              topic.metadata.refresh.interval.ms
    queue.buffering.max.ms    queue.buffering.max.messages  queue.enqueue.timeout.ms
    batch.num.messages        send.buffer.bytes             client.id
    broker.list
  ]

  attr_reader :producer, :send_method, :options

  # Create a Kafka Producer
  #
  # options:
  # metadata_broker_list: ["localhost:9092"] - REQUIRED: a seed list of kafka brokers
  def initialize(opts = {})
    @options = opts.reduce({}) do |opts, (k, v)|
      opts[k.to_s.gsub(/_/, '.')] = v
      opts
    end
    if options['broker.list']
      options['metadata.broker.list'] = options.delete 'broker.list'
    end
    if options['compressed.topics'].to_s == 'none'
      options.delete 'compressed.topics'
{'fraction_non_alphanumeric': 0.10691144708423327, 'fraction_numerical': 0.00734341252699784, 'mean_word_length': 2.8883291351805207, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3072779', 'n_tokens_mistral': 1496, 'n_tokens_neox': 1461, 'n_words': 303}