com.opendxl.databus.producer

Class DatabusProducer<P>

java.lang.Object

com.opendxl.databus.producer.Producer<P>

com.opendxl.databus.producer.DatabusProducer<P>

public class DatabusProducer<P>
extends Producer<P>

Publishes records to Databus.

The producer is thread safe. Sharing a single producer instance across threads will generally be faster than having multiple instances.

Here is a simple example of using the DatabusProducer to send records.

 
 // Prepare DatabusProducer configuration
 Map config = new HashMap<String,Object>();
 config.put("bootstrap.servers", "localhost:9092");
 config.put("client.id", "DemoProducer");

 // Create a RoutingData(String, String, String)
  getInstance
 String topic = "topic1";
 String shardingKey = "messageKey"; // optional
 String tenantGroup = group0        // optional
 RoutingData routeInformation = new RoutingData(topic, shardingKey, tenantGroup);

 // Create a Headers getInstance
 Headers headers = new Headers();
 headers.put(HeadersField.SOURCE_ID, "23452145-23452435-3245432");
 headers.put(HeadersField.TENANT_ID, "578-790-870-363265");

 // Create a MessagePayload getInstance
 String message = "Hello World";
 MessagePayload<byte[]> payload = new MessagePayload<>(message.getBytes());

 // Create a DatabusProducer getInstance
 Producer<byte[]> producer = new DatabusProducer<>(config, new ByteArraySerializer());

 // Create a ProducerRecord getInstance
 ProducerRecord<byte[]> record = new ProducerRecord<>(routeInformation, headers,payload)

 try {
    RecordMetadata result = producer.send(record, new ProducerCallback());
 } catch (Exception e) {
    e.printStackTrace();
 }


 // Implements a Callback class
 public class ProducerCallback implements Callback {

    // A callback method the user can implement to provide asynchronous handling of request completion.
    This method will
    // be called when the record sent to the server has been acknowledged. Exactly one of the arguments will be
    // non-null.
    // @param metadata  The metadata for the record that was sent (i.e. the partition and offset). Null if an error
    //           occurred.
    // @param exception The exception thrown during processing of this record. Null if no error occurred.
    public void onCompletion(RecordMetadata metadata, Exception exception) {
       System.out.println(
           "MSG SENT  TOPICS:"+ metadata.topic() + " PARTITION:" + metadata.partition() +
           " OFFSET:" + metadata.offset() );
      }
 }
 
 

Constructor Summary

Constructors

Constructor and Description
DatabusProducer(java.util.Map<java.lang.String,java.lang.Object> configs, Serializer<P> messageSerializer) A DatabusProducer is instantiated by providing a set of key-value as configuration.
DatabusProducer(java.util.Map<java.lang.String,java.lang.Object> configs, Serializer<P> messageSerializer, Credential credential) A DatabusProducer is instantiated by providing a set of key-value as configuration.
DatabusProducer(java.util.Properties properties, Serializer<P> messageSerializer) A producer is instantiated by providing a set of key-value as configuration.
DatabusProducer(java.util.Properties properties, Serializer<P> messageSerializer, Credential credential) A producer is instantiated by providing a set of key-value as configuration.

Method Summary

Methods inherited from class com.opendxl.databus.producer.Producer

abortTransaction, beginTransaction, close, close, commitTransaction, flush, getConfiguration, initTransactions, metrics, partitionsFor, recordBatchSizeAvgMetric, recordBatchSizeMaxMetric, recordByteRatePerTopicMetric, recordByteTotalPerTopicMetric, recordErrorRateMetric, recordErrorRatePerTopicMetric, recordErrorTotalMetric, recordErrorTotalPerTopicMetric, recordSendRateMetric, recordSendRatePerTopicMetric, recordSendTotalMetric, recordSendTotalPerTopicMetric, recordSizeAvgMetric, recordSizeMaxMetric, send, send, sendOffsetsToTransaction, setConfiguration

Methods inherited from class java.lang.Object

equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

DatabusProducer

public DatabusProducer(java.util.Map<java.lang.String,java.lang.Object> configs,
                       Serializer<P> messageSerializer)

A DatabusProducer is instantiated by providing a set of key-value as configuration. Valid configuration strings are documented here. Values can be either strings or Objects of the appropriate type (for example a numeric configuration would accept either the string "42" or the integer 42).

Specific key-value from ProducerDefaultConfiguration will be added unless they are provided by SDK user

Parameters:

configs - The DatabusProducer configs

messageSerializer - an instance of MessageSerializer provided by SDK's user, used to transform the message to byte[] when Producer.send(ProducerRecord, Callback) sends the message

Throws:

DatabusClientRuntimeException - if a DatabusProducer instance was not able to be created

DatabusProducer

public DatabusProducer(java.util.Map<java.lang.String,java.lang.Object> configs,
                       Serializer<P> messageSerializer,
                       Credential credential)

A DatabusProducer is instantiated by providing a set of key-value as configuration. Valid configuration strings are documented here. Values can be either strings or Objects of the appropriate type (for example a numeric configuration would accept either the string "42" or the integer 42).

Specific key-value from ProducerDefaultConfiguration will be added unless they are provided by SDK user

Parameters:

configs - The DatabusProducer configs

messageSerializer - an instance of MessageSerializer provided by SDK's user, used to transform the message to byte[] when Producer.send(ProducerRecord, Callback) sends the message

credential - identity to authentication/authorization

Throws:

DatabusClientRuntimeException - if a DatabusProducer instance was not able to be created

DatabusProducer

public DatabusProducer(java.util.Properties properties,
                       Serializer<P> messageSerializer)

A producer is instantiated by providing a set of key-value as configuration. Valid configuration strings are documented here.

Specific key-value from ProducerDefaultConfiguration will be added unless they are provided by SDK user

Parameters:

properties - The DatabusProducer configs

messageSerializer - an instance of MessageSerializer provided by SDK's user, used to transform the message to byte[] when Producer.send(ProducerRecord, Callback) sends the message

Throws:

DatabusClientRuntimeException - if a DatabusProducer instance was not able to be created

DatabusProducer

public DatabusProducer(java.util.Properties properties,
                       Serializer<P> messageSerializer,
                       Credential credential)

A producer is instantiated by providing a set of key-value as configuration. Valid configuration strings are documented here.

Specific key-value from ProducerDefaultConfiguration will be added unless they are provided by SDK user

Parameters:

properties - The DatabusProducer configs

messageSerializer - a instance of MessageSerializer provided by SDK's user, used to transform the message to byte[] when Producer.send(ProducerRecord, Callback) sends the message

credential - identity to authenticate/authorization

Throws:

DatabusClientRuntimeException - if a DatabusProducer instance was not able to be created