Class: Kafka::KafkaProducer

Inherits:
Object
  • Object
show all
Defined in:
lib/jruby-kafka/kafka-producer.rb

Overview

noinspection JRubyStringImportInspection

Defined Under Namespace

Classes: RubyCallback

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(opts = {}) ⇒ KafkaProducer

Create a Kafka producer.

For other configuration properties and their default values see kafka.apache.org/documentation.html#producerconfigs and kafka.apache.org/0100/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html.

Parameters:

  • config (Hash) —

    the producer configuration.



35
36
37
38
 
# File 'lib/jruby-kafka/kafka-producer.rb', line 35

def initialize(opts = {})
  @properties = opts.clone
  super Kafka::Utility.java_properties @properties
end

Instance Attribute Details

#properties ⇒ Object (readonly)

Returns the value of attribute properties.



21
22
23
 
# File 'lib/jruby-kafka/kafka-producer.rb', line 21

def properties
  @properties
end

Instance Method Details

#send_msg(topic, partition, key, value, timestamp = nil, &block) ⇒ Object

Send a message to the cluster.

Parameters:

  • topic (String) —

    The topic to send the message to.

  • partition (Integer, nil) —

    The topic partition to send the message to, or nil to allow the configured partitioner class to select the partition.

  • key (String, nil) —

    The message key, if there is one. Otherwise, nil.

  • value (String) —

    The message value.

  • timestamp (Integer, nil) (defaults to: nil) —

    The message timestamp in milliseconds. If nil, the producer will assign it the current time.

Raises:

  • (FailedToSendMessageException) —

    if it can’t send the message



54
55
56
57
58
59
60
61
62
 
# File 'lib/jruby-kafka/kafka-producer.rb', line 54

def send_msg(topic, partition, key, value, timestamp=nil, &block)
  record = ProducerRecord.new(topic, partition, timestamp, key, value)

  if block
    send_cb_method record, RubyCallback.new(block)
  else
    send_method record
  end
end