org.voltdb.client.topics

Class VoltDBKafkaPartitioner

java.lang.Object

org.apache.kafka.clients.producer.internals.DefaultPartitioner

org.voltdb.client.topics.VoltDBKafkaPartitioner

All Implemented Interfaces:

java.io.Closeable, java.lang.AutoCloseable, org.apache.kafka.clients.producer.Partitioner, org.apache.kafka.common.Configurable

public class VoltDBKafkaPartitioner
extends org.apache.kafka.clients.producer.internals.DefaultPartitioner

VoltDBKafkaPartitioner is a partitioner to calculate VoltDB partition id from the key value. Here is a simple example of using the producer to send records with VoltDBKafkaPartitioner

 
 Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("bootstrap.servers.voltdb", "localhost:21212");
 props.put("acks", "all");
 props.put("retries", 0);
 props.put("batch.size", 16384);
 props.put("linger.ms", 1);
 props.put("buffer.memory", 33554432);
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, VoltDBKafkaPartitioner.class.getName());
 Producer<String, String> producer = new KafkaProducer<>(props);
 for (int i = 0; i < 100; i++) {
     producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));
 }
 producer.close();
 
 

The bootstrap.servers.voltdb is required to calculate partition id from the key.

Kafka client properties used by the partitioner

• ProducerConfig.BOOTSTRAP_SERVERS_CONFIG - If bootstrap.servers.voltdb is not set
• CommonClientConfigs.SECURITY_PROTOCOL_CONFIG
• SaslConfigs.SASL_MECHANISM - Must be PLAIN
• SaslConfigs.SASL_JAAS_CONFIG
• SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG
• SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG

Field Summary

Fields

Modifier and Type	Field and Description
static java.lang.String	BOOTSTRAP_SERVERS_VOLTDB Configuration for a VoltDB client to connect to VoltDB cluster: comma separated list of the form server[:port]
protected org.voltdb.client.ClientImpl	m_client

Constructor Summary

Constructors

Constructor and Description
VoltDBKafkaPartitioner()

Method Summary

Modifier and Type	Method and Description
void	close()
void	configure(java.util.Map<java.lang.String,?> original)
protected ClientConfig	createClientConfig(org.voltdb.client.topics.VoltDBKafkaPartitioner.PartitionConfig configs) Create ClientConfig for client connection to VoltDB cluster
protected void	loadTopics()
int	partition(java.lang.String topic, java.lang.Object key, byte[] keyBytes, java.lang.Object value, byte[] valueBytes, org.apache.kafka.common.Cluster cluster) Use DefaultPartitioner for opaque topics, otherwise use VoltDB hash mechanism for partition calculation.

Methods inherited from class org.apache.kafka.clients.producer.internals.DefaultPartitioner

onNewBatch, partition

Methods inherited from class java.lang.Object

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

Field Detail

BOOTSTRAP_SERVERS_VOLTDB

public static final java.lang.String BOOTSTRAP_SERVERS_VOLTDB

Configuration for a VoltDB client to connect to VoltDB cluster: comma separated list of the form server[:port]

See Also:

Constant Field Values

m_client

protected org.voltdb.client.ClientImpl m_client

Constructor Detail

VoltDBKafkaPartitioner

public VoltDBKafkaPartitioner()

Method Detail

configure

public void configure(java.util.Map<java.lang.String,?> original)

Specified by:

configure in interface org.apache.kafka.common.Configurable

Overrides:

configure in class org.apache.kafka.clients.producer.internals.DefaultPartitioner

partition

public int partition(java.lang.String topic,
                     java.lang.Object key,
                     byte[] keyBytes,
                     java.lang.Object value,
                     byte[] valueBytes,
                     org.apache.kafka.common.Cluster cluster)

Use DefaultPartitioner for opaque topics, otherwise use VoltDB hash mechanism for partition calculation. Under the following corner cases that the partitioner could direct a producer to wrong partitions:

• When opaque topics are dropped, then recreated as non-opaque topics, or vice versa
• When a VoltDB cluster is elastically expanded or reduced, the hash mechanism on VoltDB client is not promptly updated.

Specified by:

partition in interface org.apache.kafka.clients.producer.Partitioner

Overrides:

partition in class org.apache.kafka.clients.producer.internals.DefaultPartitioner

close

public void close()

Specified by:

close in interface java.io.Closeable

Specified by:

close in interface java.lang.AutoCloseable

Specified by:

close in interface org.apache.kafka.clients.producer.Partitioner

Overrides:

close in class org.apache.kafka.clients.producer.internals.DefaultPartitioner

loadTopics

protected void loadTopics()

createClientConfig

protected ClientConfig createClientConfig(org.voltdb.client.topics.VoltDBKafkaPartitioner.PartitionConfig configs)

Create ClientConfig for client connection to VoltDB cluster

Parameters:

configs - Configuration properties from KafkaProducer

Returns:

ClientConfig