org.nuxeo.lib.stream.log.kafka

Class KafkaLogTailer<M extends Externalizable>

java.lang.Object

org.nuxeo.lib.stream.log.kafka.KafkaLogTailer<M>

All Implemented Interfaces:

AutoCloseable, org.apache.kafka.clients.consumer.ConsumerRebalanceListener, LogTailer<M>

public class KafkaLogTailer<M extends Externalizable>
extends Object
implements LogTailer<M>, org.apache.kafka.clients.consumer.ConsumerRebalanceListener

Since:

9.3

Field Summary

Fields

Modifier and Type	Field and Description
protected boolean	closed
protected org.apache.kafka.clients.consumer.KafkaConsumer<String,org.apache.kafka.common.utils.Bytes>	consumer
protected String	group
protected String	id
protected boolean	isRebalanced
protected Map<org.apache.kafka.common.TopicPartition,Long>	lastCommittedOffsets
protected Map<org.apache.kafka.common.TopicPartition,Long>	lastOffsets
protected RebalanceListener	listener
protected Collection<String>	names
protected Collection<LogPartition>	partitions
protected String	prefix
protected Queue<org.apache.kafka.clients.consumer.ConsumerRecord<String,org.apache.kafka.common.utils.Bytes>>	records
protected Collection<org.apache.kafka.common.TopicPartition>	topicPartitions

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	KafkaLogTailer(String prefix, String group, Properties consumerProps)

Method Summary

Modifier and Type	Method and Description
Collection<LogPartition>	assignments() Returns the list of Log name, partitions tuples currently assigned to this tailer.
protected static String	buildId(String group, Collection<LogPartition> partitions)
protected static String	buildSubscribeId(String group, Collection<String> names)
void	close()
boolean	closed() Returns true if the tailer has been closed.
void	commit() Commit current positions for all partitions (last message offset returned by read).
LogOffset	commit(LogPartition partition) Commit current position for the partition.
static <M extends Externalizable> KafkaLogTailer<M>	createAndAssign(String prefix, Collection<LogPartition> partitions, String group, Properties consumerProps)
static <M extends Externalizable> KafkaLogTailer<M>	createAndSubscribe(String prefix, Collection<String> names, String group, Properties consumerProps, RebalanceListener listener)
protected String	getNameForTopic(String topic)
String	group() Returns the consumer group.
protected M	messageOf(org.apache.kafka.common.utils.Bytes value)
void	onPartitionsAssigned(Collection<org.apache.kafka.common.TopicPartition> newPartitions)
void	onPartitionsRevoked(Collection<org.apache.kafka.common.TopicPartition> partitions)
protected int	poll(Duration timeout)
LogRecord<M>	read(Duration timeout) Read a message from assigned partitions within the timeout.
void	reset() Reset all committed positions for this group, next read will be done from beginning.
void	reset(LogPartition partition) Reset the committed position for this group on this partition, next read for this partition will be done from the beginning.
void	seek(LogOffset offset) Set the current position for a single partition.
void	toEnd() Set the current positions to the end of all partitions.
void	toLastCommitted() Set the current positions to previously committed positions.
protected long	toLastCommitted(org.apache.kafka.common.TopicPartition topicPartition)
void	toStart() Set the current positions to the beginning of all partitions.
String	toString()

Methods inherited from class java.lang.Object

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

Field Detail

group

protected final String group

prefix

protected final String prefix

lastOffsets

protected final Map<org.apache.kafka.common.TopicPartition,Long> lastOffsets

lastCommittedOffsets

protected final Map<org.apache.kafka.common.TopicPartition,Long> lastCommittedOffsets

records

protected final Queue<org.apache.kafka.clients.consumer.ConsumerRecord<String,org.apache.kafka.common.utils.Bytes>> records

consumer

protected org.apache.kafka.clients.consumer.KafkaConsumer<String,org.apache.kafka.common.utils.Bytes> consumer

id

protected String id

topicPartitions

protected Collection<org.apache.kafka.common.TopicPartition> topicPartitions

partitions

protected Collection<LogPartition> partitions

closed

protected boolean closed

names

protected Collection<String> names

listener

protected RebalanceListener listener

isRebalanced

protected boolean isRebalanced

Constructor Detail

KafkaLogTailer

protected KafkaLogTailer(String prefix,
                         String group,
                         Properties consumerProps)

Method Detail

createAndAssign

public static <M extends Externalizable> KafkaLogTailer<M> createAndAssign(String prefix,
                                                                           Collection<LogPartition> partitions,
                                                                           String group,
                                                                           Properties consumerProps)

createAndSubscribe

public static <M extends Externalizable> KafkaLogTailer<M> createAndSubscribe(String prefix,
                                                                              Collection<String> names,
                                                                              String group,
                                                                              Properties consumerProps,
                                                                              RebalanceListener listener)

buildId

protected static String buildId(String group,
                                Collection<LogPartition> partitions)

buildSubscribeId

protected static String buildSubscribeId(String group,
                                         Collection<String> names)

read

public LogRecord<M> read(Duration timeout)
                  throws InterruptedException

Description copied from interface: LogTailer

Read a message from assigned partitions within the timeout.

Specified by:

read in interface LogTailer<M extends Externalizable>

Returns:

null if there is no message in the queue after the timeout.

Throws:

InterruptedException

getNameForTopic

protected String getNameForTopic(String topic)

messageOf

protected M messageOf(org.apache.kafka.common.utils.Bytes value)

poll

protected int poll(Duration timeout)
            throws InterruptedException

Throws:

InterruptedException

toEnd

public void toEnd()

Description copied from interface: LogTailer

Set the current positions to the end of all partitions.

Specified by:

toEnd in interface LogTailer<M extends Externalizable>

toStart

public void toStart()

Description copied from interface: LogTailer

Set the current positions to the beginning of all partitions.

Specified by:

toStart in interface LogTailer<M extends Externalizable>

toLastCommitted

public void toLastCommitted()

Description copied from interface: LogTailer

Set the current positions to previously committed positions.

Specified by:

toLastCommitted in interface LogTailer<M extends Externalizable>

toLastCommitted

protected long toLastCommitted(org.apache.kafka.common.TopicPartition topicPartition)

seek

public void seek(LogOffset offset)

Description copied from interface: LogTailer

Set the current position for a single partition. Do not change other partitions positions.

Specified by:

seek in interface LogTailer<M extends Externalizable>

reset

public void reset()

Description copied from interface: LogTailer

Reset all committed positions for this group, next read will be done from beginning.

Specified by:

reset in interface LogTailer<M extends Externalizable>

reset

public void reset(LogPartition partition)

Description copied from interface: LogTailer

Reset the committed position for this group on this partition, next read for this partition will be done from the beginning.

Specified by:

reset in interface LogTailer<M extends Externalizable>

commit

public void commit()

Description copied from interface: LogTailer

Commit current positions for all partitions (last message offset returned by read).

Specified by:

commit in interface LogTailer<M extends Externalizable>

commit

public LogOffset commit(LogPartition partition)

Description copied from interface: LogTailer

Commit current position for the partition.

Specified by:

commit in interface LogTailer<M extends Externalizable>

Returns:

the committed offset, can return null if there was no previous read done on this partition.

assignments

public Collection<LogPartition> assignments()

Description copied from interface: LogTailer

Returns the list of Log name, partitions tuples currently assigned to this tailer. Assignments can change only if the tailer has been created using LogManager.subscribe(java.lang.String, java.util.Collection<java.lang.String>, org.nuxeo.lib.stream.log.RebalanceListener).

Specified by:

assignments in interface LogTailer<M extends Externalizable>

group

public String group()

Description copied from interface: LogTailer

Returns the consumer group.

Specified by:

group in interface LogTailer<M extends Externalizable>

closed

public boolean closed()

Description copied from interface: LogTailer

Returns true if the tailer has been closed.

Specified by:

closed in interface LogTailer<M extends Externalizable>

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface LogTailer<M extends Externalizable>

toString

public String toString()

Overrides:

toString in class Object

onPartitionsRevoked

public void onPartitionsRevoked(Collection<org.apache.kafka.common.TopicPartition> partitions)

Specified by:

onPartitionsRevoked in interface org.apache.kafka.clients.consumer.ConsumerRebalanceListener

onPartitionsAssigned

public void onPartitionsAssigned(Collection<org.apache.kafka.common.TopicPartition> newPartitions)

Specified by:

onPartitionsAssigned in interface org.apache.kafka.clients.consumer.ConsumerRebalanceListener