Kafka

Syntax

producer(config, [errCallback])

Details

Create a Kafka producer based on the specified configurations and return the handle.

Note that there is currently no explicit function to close the producer, including the close function in DolphinDB server. To close the producer, set it to null to trigger the destructor.

Parameters

• config: A dictionary indicating the Kafka producer configuration, whose key is a string and value is a string or a boolean. Generally, the configuration of a producer only requires the cluster address for connection. If the cluster enables security authentication, additional configurations are required. For details, refer to Kafka Configuration.
• errCallback (optional): A function to be called when an error or warning occurs in the background. This function accepts three STRING parameters: level ("ERROR" or "WARNING"), errorName, and message (specific error information).

Configuration Parameters

Configuration Parameters for Kafka Producer

Examples

• Create a local Kafka producer:

  producerCfg = dict(STRING, ANY)
  producerCfg["metadata.broker.list"] = "localhost:9092"
  handle = kafka::producer(producerCfg)

• Create a Kafka producer with a callback function:

  producerCfg = dict(string, any)
  producerCfg["bootstrap.servers"] = "192.168.100.3"
  share table(1:0, `level`err`reason, [STRING,STRING,STRING]) as produceDest
  def producerCallback(mutable dest, level, err, reason) {
      tb = table([level] as level, [err] as err, [reason] as reason)
      dest.append!(tb)
  }
  producer = kafka::producer(producerCfg, producerCallback{produceDest})

Syntax

produce(producer, topic, key, value, marshalType, [partition])

Details

Send messages to the specified Kafka topic using the given format. For detailed usage and consumer integration, refer to Usage Examples.

Note that:

• When sending tabular data in JSON format, the data in a column will be written into an array.
• Data serialized in DolphinDB format can only be received using the DolphinDB kafka plugin with marshalType specified as "DOLPHINDB"; otherwise, the binary serialized data cannot be processed, which may lead to garbled data.

Parameters

• producer: A Kafka producer handle.
• topic: A STRING scalar indicating the Kafka topic to send messages to.
• key: A STRING scalar indicating the key of the Kafka message. Messages with the same key will be sent to the same partition.
• value: A STRING scalar indicating the data of the Kafka message.
• marshalType (optional): A STRING scalar specifying the serialization format. It can be:
  • "JSON": Serialization in JSON format.
  • "DOLPHINDB": Serialization in DolphinDB internal format.
  • "PLAIN": Plain data with no serialization. This option only supports sending strings.
• partition (optional): An integer indicating the partition number of the Kafka topic to send messages to. If unspecified, the cluster will evenly distribute the messages based on key.

Syntax

consumer(config, [errCallback])

Details

Create a Kafka consumer based on the specified configurations and return the handle.

Note that there is currently no explicit function to close the consumer, including the close function in DolphinDB server. To close the consumer, set it to null to trigger the destructor. If the consumer to be closed is used in createSubJob, the corresponding background job also needs to be canceled.

Parameters

• config: A dictionary indicating the Kafka consumer configuration, whose key is a string and value is a string or a boolean. Parameters metadata.broker.list and group.id are required for consumer configuration. For descriptions of general configuration parameters, refer to producer.
• errCallback (optional): A function to be called when an error or warning occurs in the background. This function accepts three STRING parameters: level ("ERROR" or "WARNING"), errorName, and message (specific error information).

Configuration Parameters for Kafka Consumer

Examples

Create a local Kafka consumer and specify group.id as "group1":

consumerCfg = dict(STRING, ANY)
consumerCfg["group.id"] = "group1"
consumerCfg["metadata.broker.list"] = "localhost:9092";
consumer = kafka::consumer(consumerCfg)

Syntax

subscribe(consumer, topics)

Details

Subscribe to a Kafka topic and return no value.

Note that:

• The number of consumers in the same group cannot exceed the number of partitions in the topic. Otherwise, unassigned consumers will be unable to participate in the rebalance, even if assigned consumers exit.

• A failed subscription or a subscription to a non-existent topic returns no information or log. To access the subscription status, use the getAssignment method.

Parameters

• consumer: A Kafka consumer handle.
• topics: A STRING vector indicating the topics to subscribe to.

Syntax

consumerPoll(consumer, [timeout=1000], [marshalType])

Details

Save the subscribed data to DolphinDB and return a tuple.

The first element is a string indicating the error message, which will be empty if consumerPoll succeeds. The second element is a tuple including the following elements: topic, partition, key, value, and the timestamp when the consumer received the data. For detailed usage and consumer integration, refer to Usage Examples.

Parameters

• consumer: A Kafka consumer handle.
• timeout (optional): The maximum amount of time in milliseconds to wait for a polling. The default value is 1000 milliseconds.
• marshalType (optional): A STRING scalar specifying the parsing format. If unspecified, consumerPoll detects the subscribed data and parses it in JSON or DOLPHINDB format based on the detection results. It can be:
  • "JSON": Parse in JSON format.
  • "DOLPHINDB": Parse in DolphinDB internal format.
  • "PLAIN": Unparsed plain data.

Syntax

consumerPollBatch(consumer, batchSize, [timeout=1000], [marshalType])

Details

Save the subscribed data to DolphinDB in batch and return a tuple.

Each element of the output is a tuple including the following elements: topic, partition, key, value, and the timestamp when the consumer received the data. For detailed usage and consumer integration, refer to Usage Examples.

Parameters

• consumer: A Kafka consumer handle.
• batch_size: An integer indicating the number of messages you want to get.
• timeout (optional): An integer indicating the maximum amount of time in milliseconds to get messages.
• marshalType (optional): A STRING scalar specifying the parsing format. If unspecified, consumerPollBatch detects the subscribed data and parses it in JSON or DolphinDB format based on the detection results. It can be:
  • "JSON": Parse in JSON format.
  • "DOLPHINDB": Parse in DolphinDB internal format.
  • "PLAIN": Unparsed plain data.

Syntax

createSubJob(consumer, table, parser, actionName, [throttle=1.0], [autoCommit], [msgAsTable=false], [batchSize=1], [queueDepth=1000000])

Details

Return a subJob connection handle. For detailed usage, refer to Usage Examples.

Parameters

• consumer: A Kafka consumer handle.
• table: A table to store the subscribed messages. It can be a shared stream table, shared in-memory table, shared keyed table, shared indexed table, or DFS table.
• parser: A unary, binary, or ternary function to process the subscribed messages and insert the result into the table parameter.
  • If msgAsTable is false, parser accepts 1-3 STRING parameters as the input: the first is the message value, the second the key, and the third the topic.
  • If msgAsTable is true, parser accepts a table with three STRING type columns: value, key, and topic.
• actionName: A string indicating the name of the subscription job, which must not duplicate existing names.
• throttle (optional): A floating-point number in seconds indicating how long to wait before the parser processes the incoming messages if the batchSize condition has not been reached. The default value is 1.0. It is only effective when msgAsTable is true.
• autoCommit (optional): A boolean indicating whether to perform an automatic commit after the parser processes the messages and inserts the result into table.
  • If enable.auto.commit is false for the specified consumer, setting autoCommit to true can prevent data loss cause by system crashes, as the uncommitted data will be re-received upon system recovery.
  • If enable.auto.commit is false while autoCommit is unspecified or set to true, automatic commits will be performed after the messages have been processed.
• msgAsTable (optional): A boolean indicating whether the subscribed data is ingested into parser as a table. The default value is false.
• batchSize (optional): An integer indicating the number of unprocessed messages to trigger the parser. The default value is 0. It is only effective when msgAsTable is true.
  • If it is positive, messages will be processed when the number of unprocessed messages reaches batchSize.
  • If it is unspecified or non-positive, messages will be processed immediately as they come in.
• queueDepth (optional): A positive integer indicating the depth of the receiving queue in Kafka’s background. The default value is 1000000.

Syntax

getJobStat()

Details

Get the statuses of the subscription jobs created by createSubJob and return a table.

Syntax

cancelSubJob(handle|actionName)

Details

Cancel the specified subscription job in the background. The canceled job can no longer be retrieved by getJobStat.

Note: To maintain compatibility, this method also accepts the subscriptionId returned by getJobStat. If actionName and subscriptionId share the same name, the subscription job specified by actionName will be canceled preferentially.

Parameters

• handle|actionName: The result of createSubJobor the actionName specified by it.

Syntax

getSubJobConsumer(handle|actionName)

Details

Get the consumer of the specified subscription job in the background created by createSubJob.

Parameters

• handle|actionName: The result of createSubJobor the actionName specified by it.

Syntax

commit(consumer, [topics], [partitions], [offsets])

Details

Synchronously commit the offset of the latest consumed messages to the Kafka cluster.

Parameters

• consumer: A Kafka consumer handle.
• topics (optional): A STRING scalar or vector indicating the subscribed topic(s).
• partitions (optional): An INT scalar or vector indicating the subscribed partition(s).
• offsets (optional): An INT scalar or vector indicating the offset(s) for each topic.

Note:

• topics, partitions, and offsets must have the same format and length, with their elements corresponding to each other.
• If they are specified, only specified offsets will be commited.
• If unspecified, the latest consumption offsets for all subscribed partitions will be committed.

Syntax

assign(consumer, topics, partitions, offsets)

Details

Unlike the method subscribe, assign enables you to assign specific topics, partitions, and offsets to the consumer.

Parameters

• consumer: A Kafka consumer handle.
• topics: A STRING scalar or vector indicating the subscribed topic(s).
• partitions: An INT scalar or vector indicating the subscribed partition(s).
• offsets: An INT scalar or vector indicating the offset(s) for each topic.

Note that:

• topics, partitions, and offsets must have the same format and length, with their elements corresponding to each other.
• The data in the same partition of a topic can be consumed by multiple consumers using assign.
• To avoid undefined behaviors, it is recommended not to apply subscribe and assign to the same consumer or execute these methods multiple times.

Examples

Assign the messages in the partition "1" of the "test" topic with the offset 3000 to the consumer.

kafka::assign(consumer, "test", 1, 3000)

Syntax

unassign(consumer)

Details

Cancel the specified consumer’s subscription to all topics. This method applies to subscriptions created by both subscribe and assign.

Parameters

• consumer: A Kafka consumer handle.

Syntax

getAssignment(consumer)

Details

Get the assignment information of the specified consumer. Return a table with 3 columns: topic, partition, and offset.

Note: If the consumer only calls subscribe and does not call assign, this method can also return the specific assignment information.

Parameters

• consumer: A Kafka consumer handle.

Syntax

getOffsetInfo(consumer, topic, partition)

Details

Get the offset information about the specified topics and partitions. Return a table with the following fields:

Parameters

• consumer: A Kafka consumer handle.
• topics: A STRING scalar or vector indicating the subscribed topic(s).
• partitions: An INT scalar or vector indicating the subscribed partition(s).

Note that:

• topics and partitions must have the same format and length, with their elements corresponding to each other.
• The specified topics and partitions must be subscribed by the consumer.

Examples

The following example demonstrates the differences between getAssignment and getOffsetInfo.

• Send 10 messages to the topic "offsetTopic".

  producerCfg = dict(string, any);
  producerCfg["bootstrap.servers"] = "localhost"
  producer = kafka::producer(producerCfg)
  for (i in 0..9) {
      kafka::produce(producer, "offsetTopic", "0", "message", "PLAIN")
  }

• Create a consumer, which consumes the subscribed data twice.

  consumerCfg = dict(string, any);
  consumerCfg["metadata.broker.list"] = "localhost"
  consumerCfg["enable.auto.commit"] = "false"
  consumerCfg["group.id"] = "test"
  consumer = kafka::consumer(consumerCfg)
  kafka::assign(consumer, "offsetTopic", 0, 3)
  kafka::consumerPoll(consumer,,"PLAIN")
  kafka::consumerPoll(consumer,,"PLAIN")

• Apply getAssignment to get the assignment information about the consumer.

  kafka::getAssignment(consumer)

Output:

In the output, offset is -1,000, indicating that no specific offset is specified.

• Apply getOffsetInfo to get the offset information about the partition 0 in "offsetTopic".

  kafka::getOffsetInfo(consumer, "offsetTopic", 0)

  Output:

  topic	partition	minOffset	maxOffset	offsetPosition	offsetCommitted
  offsetTopic	0	0	10	2	-1,001

  maxOffset is 10, indicating that 10 messages are sent. offsetPosition is 2, indicating that the consumer consumes 2 messages. Since enable.auto.commit is set to false, offsetCommitted is an invalid value, indicating that no offsets have been committed.

• Apply getOffsetInfo again after committing the offsets using commit.

  kafka::commit(consumer)
  kafka::getOffsetInfo(consumer, "offsetTopic", 0)

  Output:

  topic	partition	minOffset	maxOffset	offsetPosition	offsetCommitted
  offsetTopic	0	0	10	2	2

  offsetCommitted becomes 2, indicating that 2 offsets have been committed.

Syntax

getMemId(consumer)

Details

Each consumer has a unique ID. This method can be used to get the ID of the specified consumer and return a string.

Parameters

• consumer: A Kafka consumer handle.

Examples

consumerCfg = dict(STRING, ANY)
consumerCfg["group.id"] = string(now())
consumerCfg["metadata.broker.list"] = "localhost:9092";
consumer = kafka::consumer(consumerCfg)
kafka::subscribe(consumer, "test");
kafka::consumerPoll(consumer); // Without this step, the consumer may not be assigned an ID in the Kafka cluster
kafka::getMemId(consumer)
// output: 'rdkafka-d9eded44-358f-49fc-be01-cc099b121d59'

Syntax

getMetadata(handle|conf)

Details

Get the metadata of the specified Kafka cluster. Return a dictionary with the following key-value pairs:

Parameters

• handle|conf: It can be the handle of an existing producer or consumer, a dictionary used to establish a connection, or the IP address and port number of a Kafka cluster.

Examples

Get the metadata of the Kafka cluster with the IP address and port number "localhost:9092".

kafka::getMetadata("localhost:9092")

Get the metadata of the Kafka cluster corresponding to the specified consumer handle.

kafka::getMetadata(consumer)

Output:

brokers->
  id host           port
  -- -------------- ----
  2  192.168.100.45 9092
  1  192.168.100.44 9092
  0  192.168.100.43 9092
consumerGroup->
  test->
    state->Stable
    protocolType->consumer
    protocol->range
    error->Success
    members->
      rdkafka-0b501728-5344-4f2f-a450-bc056f6e3400->
        clientID->rdkafka
        clientHost->/192.168.0.38
        memberAssignmentVersion->0
        partitions->
          topic partition offset
          ----- --------- ------
          test  0         -1001 
          test  1         -1001 
          test  2         -1001 
topics->
  test->
    id error   leader replicas
    -- ------- ------ --------
    0  Success 0      [0]     
    1  Success 1      [1]     
    2  Success 2      [2]