cn.leancloud.kafka.consumer

Class LcKafkaConsumerBuilder<K,V>

java.lang.Object

cn.leancloud.kafka.consumer.LcKafkaConsumerBuilder<K,V>

public final class LcKafkaConsumerBuilder<K,V>
extends Object

Method Summary

Modifier and Type	Method and Description
<K1 extends K,V1 extends V> LcKafkaConsumer<K1,V1>	buildAsync()
<K1 extends K,V1 extends V> LcKafkaConsumer<K1,V1>	buildAuto() Build a consumer which commit offset automatically in fixed interval.
<K1 extends K,V1 extends V> LcKafkaConsumer<K1,V1>	buildPartialAsync()
<K1 extends K,V1 extends V> LcKafkaConsumer<K1,V1>	buildPartialSync()
<K1 extends K,V1 extends V> LcKafkaConsumer<K1,V1>	buildSync()
LcKafkaConsumerBuilder<K,V>	gracefulShutdownMs(Duration duration)
LcKafkaConsumerBuilder<K,V>	gracefulShutdownMs(long gracefulShutdownMs)
LcKafkaConsumerBuilder<K,V>	maxPendingAsyncCommits(int maxPendingAsyncCommits) When using async consumer to commit offset asynchronously, this argument can force consumer to do a synchronous commit after there's already maxPendingAsyncCommits async commits on the fly without response from broker.
LcKafkaConsumerBuilder<K,V>	messageHandler(MessageHandler<K,V> messageHandler) Change the MessageHandler to handle the consumed msg from kafka.
static <K,V> LcKafkaConsumerBuilder<K,V>	newBuilder(Map<String,Object> kafkaConfigs, MessageHandler<K,V> messageHandler) Create a LcKafkaConsumerBuilder used to build LcKafkaConsumer.
static <K,V> LcKafkaConsumerBuilder<K,V>	newBuilder(Map<String,Object> kafkaConfigs, MessageHandler<K,V> messageHandler, org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer, org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer) Create a LcKafkaConsumerBuilder used to build LcKafkaConsumer.
LcKafkaConsumerBuilder<K,V>	pollTimeout(Duration pollTimeout) The pollTimeout is the maximum time spent waiting in polling data from kafka broker if data is not available in the buffer.
LcKafkaConsumerBuilder<K,V>	pollTimeoutMs(long pollTimeoutMs) The pollTimeout is the maximum time spent waiting in polling data from kafka broker if data is not available in the buffer.
LcKafkaConsumerBuilder<K,V>	workerPool(ExecutorService workerPool, boolean shutdownOnStop) The thread pool used by consumer to handle the consumed messages from kafka.

Methods inherited from class java.lang.Object

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

Method Detail

newBuilder

public static <K,V> LcKafkaConsumerBuilder<K,V> newBuilder(Map<String,Object> kafkaConfigs,
                                                           MessageHandler<K,V> messageHandler)

Create a LcKafkaConsumerBuilder used to build LcKafkaConsumer.

Parameters:

kafkaConfigs - the kafka consumer configs. Please refer this document for valid configurations.

messageHandler - a MessageHandler to handle the consumed msg from kafka

Returns:

a new LcKafkaConsumerBuilder

newBuilder

public static <K,V> LcKafkaConsumerBuilder<K,V> newBuilder(Map<String,Object> kafkaConfigs,
                                                           MessageHandler<K,V> messageHandler,
                                                           org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer,
                                                           org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer)

Create a LcKafkaConsumerBuilder used to build LcKafkaConsumer.

Parameters:

kafkaConfigs - the kafka consumer configs. Please refer this document for valid configurations.

messageHandler - a MessageHandler to handle the consumed msg from kafka

keyDeserializer - The deserializer for key that implements Deserializer

valueDeserializer - The deserializer for value that implements Deserializer

Returns:

a new LcKafkaConsumerBuilder

pollTimeoutMs

public LcKafkaConsumerBuilder<K,V> pollTimeoutMs(long pollTimeoutMs)

The pollTimeout is the maximum time spent waiting in polling data from kafka broker if data is not available in the buffer.

If 0, poll operation will return immediately with any records that are available currently in the buffer, else returns empty.

Must not be negative.

Parameters:

pollTimeoutMs - the poll timeout in milliseconds

Returns:

this

pollTimeout

public LcKafkaConsumerBuilder<K,V> pollTimeout(Duration pollTimeout)

The pollTimeout is the maximum time spent waiting in polling data from kafka broker if data is not available in the buffer.

If 0, poll operation will return immediately with any records that are available currently in the buffer, else returns empty.

Must not be negative.

Parameters:

pollTimeout - the poll timeout duration

Returns:

this

gracefulShutdownMs

public LcKafkaConsumerBuilder<K,V> gracefulShutdownMs(long gracefulShutdownMs)

gracefulShutdownMs

public LcKafkaConsumerBuilder<K,V> gracefulShutdownMs(Duration duration)

maxPendingAsyncCommits

public LcKafkaConsumerBuilder<K,V> maxPendingAsyncCommits(int maxPendingAsyncCommits)

When using async consumer to commit offset asynchronously, this argument can force consumer to do a synchronous commit after there's already maxPendingAsyncCommits async commits on the fly without response from broker.

Parameters:

maxPendingAsyncCommits - do a synchronous commit when pending async commits beyond this limit

Returns:

this

messageHandler

public LcKafkaConsumerBuilder<K,V> messageHandler(MessageHandler<K,V> messageHandler)

Change the MessageHandler to handle the consumed msg from kafka.

Parameters:

messageHandler - the handler to handle consumed msg

Returns:

this

workerPool

public LcKafkaConsumerBuilder<K,V> workerPool(ExecutorService workerPool,
                                              boolean shutdownOnStop)

The thread pool used by consumer to handle the consumed messages from kafka. Please note that if you are using auto commit consumer, this thread pool is not be used.

Parameters:

workerPool - a thread pool to handle consumed messages

shutdownOnStop - true to shutdown the input worker pool when this consumer closed

Returns:

this

buildAuto

public <K1 extends K,V1 extends V> LcKafkaConsumer<K1,V1> buildAuto()

Build a consumer which commit offset automatically in fixed interval. This consumer will not using other thread pool to handle consumed messages and will not pause any partition when after polling. This consumer is equivalent to:

 while (true) {
     final ConsumerRecords records = consumer.poll(pollTimeout);
     for (ConsumerRecord record : records) {
         handler.handleMessage(record.topic(), record.value());
     }
 }
 

Please note that this consumer requires these kafka configs must be set, otherwise IllegalArgumentException will be thrown:

1. max.poll.interval.ms
2. max.poll.records
3. auto.commit.interval.ms

Though all of these configs have default values in kafka, we still require every user to set them specifically. Because these configs is vital for using this consumer safely. You should tune them to ensure the polling thread in this consumer can at least poll once within max.poll.interval.ms during handling consumed messages to prevent itself from session timeout.

Note that if you set enable.auto.commit to false, this consumer will set it to true by itself.

Returns:

this

buildSync

public <K1 extends K,V1 extends V> LcKafkaConsumer<K1,V1> buildSync()

buildPartialSync

public <K1 extends K,V1 extends V> LcKafkaConsumer<K1,V1> buildPartialSync()

buildAsync

public <K1 extends K,V1 extends V> LcKafkaConsumer<K1,V1> buildAsync()

buildPartialAsync

public <K1 extends K,V1 extends V> LcKafkaConsumer<K1,V1> buildPartialAsync()