Consumer

Consumer<K, V> (Interface)

Package: io.axual.client.consumer

The axualClient.buildConsumer() call returns an implementation of Consumer interface.

AxualClient client = ... (1)

ConsumerConfig consumerConfig = ConsumerConfig.builder()
	.setKeySerializerClassName(...)
	...
	.build(); (2)

Processor<K, V> processor = ... (3)

Consumer<K, V> consumer = client.buildConsumer(consumerConfig); (4)

consumer.startConsuming(...); (5)
1 Build AxualClient object.
2 Build ConsumerConfigConsumerConfig object.
3 Build Processor object.
4 Build Consumer object.
5 Start consuming

Methods

Future<ConsumeFailedException> startConsuming()

Start consuming process. Returns a Future object wrapping a ConsumeFailedException

ConsumeFailedException stopConsuming()

Stop consuming and disconnect.

boolean isConsuming()

Indicates whether the consumer is still consuming, might return ConsumeFailedException if no longer consuming because of an error.

void close()

Closes the consumer.

ConsumerMessage<K, V> (Interface)

A wrapper for the consumed message passed to the Processor. Includes metadata information like partition, offset and timestamp.

Processor<K, V> (Interface)

Implement this interface to process each consumed message.

class MyProcessor implements Processor<K, V> {
    @Override
    public void processMessage(ConsumerMessage<K, V> message) {
        log.info("Processing message: Key: {}, Value: {}", message.getKey(), message.getValue());
    }
}

If using Java 8, above code can be simplified using Streams API as below:

Processor<K, V> processor = message -> log.info("Processing message: Key: {}, Value: {}", message.getKey(), message.getValue());

ConsumerStrategy

An Enum used for the consumers, the strategy is used to manage consumer behaviour.

AT_MOST_ONCE

In this strategy, consumer is guaranteed to never consume a message twice. But it might miss a message in certain situations like cluster deactivation and automatic switch.

AT_LEAST_ONCE

In this strategy, consumer is guaranteed to consume every message once. But it might consume same message more than once in certain situations like cluster deactivation and automatic switch. Consumer should be prepared to handle duplicate messages.