Kafka Consumer Adapter

The TIBCO StreamBase® Input Adapter for Apache Kafka Consumer allows the system to consume data from an Apache Kafka broker.

Each message from the broker contains the topic that the message was sent to, as well as the message, key, offset, and partition. When subscribing to a topic, you can request to start from a specific offset value, or give a timestamp value which the system determines from the closest offset to the given time and produce messages from that point forward.

When requesting offsets by timestamp, there are two values that have meaning:

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 output port for the Kafka Consumer adapter, and is always enabled. Use the data port to receive message data from the connected Apache Broker.

The default schema for the data output port is:

When a class is supplied for either message or key serialization, you must provide a class that implements the org.apache.kafka.common.serialization.Deserializer interface. The following is an example of such a class:

package com.streambase.sb.adapter.kafka;

import java.io.UnsupportedEncodingException;
import java.util.Map;

import org.slf4j.Logger;

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

public class DemoDeserializer implements org.apache.kafka.common
  .serialization.Deserializer<Tuple> {

  private static final String CONFIG_SCHEMA = "schema";
  private static final String CONFIG_CHARACTER_SET = "characterSet";
  private static final String CONFIG_USE_DEFAULT_CHARACTER_SET = 
    "useDefaultCharacterSet";
  private static final String CONFIG_LOGGER = "logger";
    
  private Schema schema;
  private String characterSet;
  private boolean useDefaultCharacterSet = true;
  private Logger logger;
    
  @Override
  public void configure(Map<String, ?> configs, boolean isKey) {
    schema = (Schema) configs.get(CONFIG_SCHEMA);
    characterSet = (String) configs.get(CONFIG_CHARACTER_SET);
    useDefaultCharacterSet = (Boolean) 
      configs.get(CONFIG_USE_DEFAULT_CHARACTER_SET);
    logger = (Logger) configs.get(CONFIG_LOGGER);
  }

  @Override
  public Tuple deserialize(String topic, byte[] data) {
    Tuple tuple = schema.createTuple();
    try {
        String tupleJSON = useDefaultCharacterSet ? new String(data) : 
          new String(data, characterSet);
        logger.info("Deserializing tuple for topic '" + 
          topic + "' from string: " + tupleJSON);
        TupleJSONUtil.setTupleFromJSONLoose(tuple, tupleJSON, "");            
      } catch (UnsupportedEncodingException | StreamBaseException e) {
          logger.error("Error deserializing topic '" + topic + 
            "': " + e.getMessage(), e);
      }
      return tuple;
  }

  @Override
  public void close() {
  }

}