Package org.nuxeo.lib.stream.log.kafka

Class KafkaLogAppender<M extends Externalizable>

java.lang.Object

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

All Implemented Interfaces:

AutoCloseable, CloseableLogAppender<M>, LogAppender<M>

public class KafkaLogAppender<M extends Externalizable>
extends Object
implements CloseableLogAppender<M>

Apache Kafka implementation of Log.

Since:

9.3

Field Summary

Fields

Modifier and Type	Field	Description
protected boolean	closed
protected Codec<M>	codec
protected Properties	consumerProps
protected Codec<M>	encodingCodec
protected Name	name
protected org.apache.kafka.clients.producer.KafkaProducer<String,org.apache.kafka.common.utils.Bytes>	producer
protected static AtomicInteger	PRODUCER_CLIENT_ID_SEQUENCE
protected Properties	producerProps
protected NameResolver	resolver
protected int	size
protected ConcurrentLinkedQueue<KafkaLogTailer<M>>	tailers
protected String	topic

Method Summary

Modifier and Type	Method	Description
LogOffset	append(int partition, String key, M message)
LogOffset	append(int partition, M message)	Append a message into a partition, returns LogOffset position of the message.
LogOffset	append(String key, M message)	Same as LogAppender.append(int, Externalizable), the queue is chosen using a hash of parameters "key".
void	close()
boolean	closed()	Returns true if the appender has been closed by the manager.
Codec<M>	getCodec()	Returns the codec used to write record.
String	getTopic()
protected boolean	isProcessed(Name group, org.apache.kafka.common.TopicPartition topicPartition, long offset)
Name	name()	Returns the Log's name.
static <M extends Externalizable> KafkaLogAppender<M>	open(Codec<M> codec, NameResolver resolver, Name name, Properties producerProperties, Properties consumerProperties)
int	size()	Returns the number of partitions in the Log.
String	toString()
boolean	waitFor(LogOffset offset, Name group, Duration timeout)	Wait for consumer to process a message up to the offset.

Methods inherited from class java.lang.Object

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

Field Detail

topic

protected final String topic

consumerProps

protected final Properties consumerProps

producerProps

protected final Properties producerProps

size

protected final int size

tailers

protected final ConcurrentLinkedQueue<KafkaLogTailer<M extends Externalizable>> tailers

name

protected final Name name

codec

protected final Codec<M extends Externalizable> codec

encodingCodec

protected final Codec<M extends Externalizable> encodingCodec

resolver

protected final NameResolver resolver

producer

protected org.apache.kafka.clients.producer.KafkaProducer<String,org.apache.kafka.common.utils.Bytes> producer

closed

protected boolean closed

PRODUCER_CLIENT_ID_SEQUENCE

protected static final AtomicInteger PRODUCER_CLIENT_ID_SEQUENCE

Method Detail

open

public static <M extends Externalizable> KafkaLogAppender<M> open(Codec<M> codec,
                                                                  NameResolver resolver,
                                                                  Name name,
                                                                  Properties producerProperties,
                                                                  Properties consumerProperties)

name

public Name name()

Description copied from interface: LogAppender

Returns the Log's name.

Specified by:

name in interface LogAppender<M extends Externalizable>

getTopic

public String getTopic()

size

public int size()

Description copied from interface: LogAppender

Returns the number of partitions in the Log.

Specified by:

size in interface LogAppender<M extends Externalizable>

append

public LogOffset append(String key,
                        M message)

Description copied from interface: LogAppender

Same as LogAppender.append(int, Externalizable), the queue is chosen using a hash of parameters "key".

Specified by:

append in interface LogAppender<M extends Externalizable>

append

public LogOffset append(int partition,
                        M message)

Description copied from interface: LogAppender

Append a message into a partition, returns LogOffset position of the message. This method is thread safe, a queue can be shared by multiple producers.

Specified by:

append in interface LogAppender<M extends Externalizable>

Parameters:

partition - index lower than LogAppender.size()

append

public LogOffset append(int partition,
                        String key,
                        M message)

waitFor

public boolean waitFor(LogOffset offset,
                       Name group,
                       Duration timeout)
                throws InterruptedException

Description copied from interface: LogAppender

Wait for consumer to process a message up to the offset. The message is processed if a consumer of the group commits a greater or equals offset. Return true if the message has been consumed, false in case of timeout.

Specified by:

waitFor in interface LogAppender<M extends Externalizable>

Throws:

InterruptedException

closed

public boolean closed()

Description copied from interface: LogAppender

Returns true if the appender has been closed by the manager.

Specified by:

closed in interface LogAppender<M extends Externalizable>

toString

public String toString()

Overrides:

toString in class Object

getCodec

public Codec<M> getCodec()

Description copied from interface: LogAppender

Returns the codec used to write record. A null codec is the default legacy encoding.

Specified by:

getCodec in interface LogAppender<M extends Externalizable>

isProcessed

protected boolean isProcessed(Name group,
                              org.apache.kafka.common.TopicPartition topicPartition,
                              long offset)

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface CloseableLogAppender<M extends Externalizable>