Kafka Producer Adapter

The TIBCO StreamBase® Output Adapter for Apache Kafka Producer allows StreamBase applications to connect to an Apache Kafka Broker and to send messages to the broker on specific topics.

Messages can be sent in various formats such as tuple, string, blob, or a custom format that you provide. The input tuple determines what the data type is for the message and key. The key for each message can also be specified as any format, just like the data.

You can provide a partition number to indicate into which partition this message should be stored, or you can leave it blank for the Kafka broker to manage partition placement. When a message is delivered to the broker, an output tuple is sent from the producer with the actual partition and offset for that message. This occurs when you enable the Enable Pass Through Fields option. The Kafka broker guarantees the message order.

The Apache Kafka adapter suite is implemented against the version of the Kafka libraries listed on the Supported Configurations page.

This section describes the properties you can set for this adapter, using the various tabs of the Properties view in StreamBase Studio.

The data port is the default input port for the Kafka Producer adapter, and is always enabled. Use the data port to send message data to the connected Apache Broker.

The default schema for the data input port is:

When you select Custom for either the message or key serialization, you must provide a class that implements the org.apache.kafka.common.serialization.Serializer interface. For example:

package com.streambase.sb.adapter.kafka;

import java.util.EnumSet;
import java.util.Map;

import org.slf4j.Logger;

import com.streambase.sb.Schema;
import com.streambase.sb.Tuple;
import com.streambase.sb.TupleJSONUtil;
import com.streambase.sb.TupleJSONUtil.Options;

public class DemoSerializer implements org.apache.kafka.common
  .serialization.Serializer<Tuple> {

  private static final String CONFIG_SCHEMA = "schema";
  private static final String CONFIG_LOGGER = "logger";
    
  private Schema schema;
  private Logger logger;
    
  @Override
  public void close() {
  }

  @Override
  public void configure(Map<String, ?> configs, boolean isKey) {
      if (configs.containsKey(CONFIG_SCHEMA)) {
          schema = (Schema)configs.get(CONFIG_SCHEMA);
      }
      if (configs.containsKey(CONFIG_LOGGER)) {
          logger = (Logger) configs.get(CONFIG_LOGGER);
      }        
  }

  @Override
  public byte[] serialize(String topic, Tuple tuple) {
      EnumSet<Options> options = EnumSet.of(Options.INCLUDE_NULL_FIELDS, Options
  .INCLUDE_NULL_FIELDS, Options.PREFER_MAP);
      String jsonTuple = TupleJSONUtil.toJSONString(tuple, options, "");
      return jsonTuple.getBytes();
  }
}