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

Class KafkaLogManager

java.lang.Object

org.nuxeo.lib.stream.log.internals.AbstractLogManager

org.nuxeo.lib.stream.log.kafka.KafkaLogManager

All Implemented Interfaces:

AutoCloseable, LogManager

public class KafkaLogManager
extends AbstractLogManager

Since:

9.3

Field Summary

Fields

Modifier and Type	Field	Description
protected final List<KafkaLogConfig>	configs
protected final KafkaLogConfig	defaultConfig
static final String	DISABLE_SUBSCRIBE_PROP
protected final Map<KafkaLogConfig,KafkaUtils>	kUtils

Fields inherited from class org.nuxeo.lib.stream.log.internals.AbstractLogManager

ADMIN_GROUP, appenders, tailers, tailersAssignments

Constructor Summary

Constructors

Constructor	Description
KafkaLogManager(String prefix, Properties producerProperties, Properties consumerProperties)
KafkaLogManager(List<KafkaLogConfig> kafkaConfigs)

Method Summary

Modifier and Type	Method	Description
protected void	checkValidPartition(LogPartition partition)
void	close()
void	create(Name name, int size)
<M extends Externalizable> CloseableLogAppender<M>	createAppender(Name name, Codec<M> codec)
boolean	delete(Name name)	Tries to delete a Log.
void	deleteConsumers()	Remove all existing consumers and their committed positions.
void	deleteRecords(Name name)	Delete all records of a stream by moving the first offsets to end of each partition.
protected <M extends Externalizable> LogTailer<M>	doCreateTailer(Collection<LogPartition> partitions, Name group, Codec<M> codec)
protected <M extends Externalizable> LogTailer<M>	doSubscribe(Name group, Collection<Name> names, RebalanceListener listener, Codec<M> codec)
boolean	exists(Name name)	Returns true if a Log with this name exists.
protected String	filterDisplayedProperties(Properties properties)
protected KafkaLogConfig	findDefaultConfig()
protected KafkaLogConfig	getConfig(Name name)
protected KafkaLogConfig	getConfig(Name name, Name group)
List<LogLag>	getLagPerPartition(Name name, Name group)	Returns the lag between consumer group and the producers for each partition.
protected int	getSize(Name name)
List<Name>	listAllNames()	Returns all the Log names.
List<Name>	listConsumerGroups(Name name)	List the consumer groups for a Log.
boolean	supportSubscribe()	Returns true if the Log LogManager.subscribe(org.nuxeo.lib.stream.log.Name, java.util.Collection<org.nuxeo.lib.stream.log.Name>, org.nuxeo.lib.stream.log.RebalanceListener, org.nuxeo.lib.stream.codec.Codec<M>) method is supported.
boolean	supportSubscribe(Name stream)	Returns true if the LogManager.subscribe(org.nuxeo.lib.stream.log.Name, java.util.Collection<org.nuxeo.lib.stream.log.Name>, org.nuxeo.lib.stream.log.RebalanceListener, org.nuxeo.lib.stream.codec.Codec<M>) method is supported for the specific Log.
String	toString()

Methods inherited from class org.nuxeo.lib.stream.log.internals.AbstractLogManager

checkInvalidAssignment, checkInvalidCodec, cleanTailers, createIfNotExists, createTailer, getAppender, getLatencyPerPartition, guessCodec, sameCodec, size, subscribe

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.nuxeo.lib.stream.log.LogManager

createIfNotExists, createTailer, createTailer, createTailer, createTailer, createTailer, createTailer, createTailer, createTailer, createTailer, createTailer, createTailer, delete, exists, getAppender, getAppender, getAppender, getLag, getLag, getLagPerPartition, getLatency, getLatency, getLatencyPerPartition, listAll, listConsumerGroups, size, subscribe, subscribe, subscribe

Field Details

DISABLE_SUBSCRIBE_PROP

public static final String DISABLE_SUBSCRIBE_PROP

See Also:

• Constant Field Values

configs

protected final List<KafkaLogConfig> configs

defaultConfig

protected final KafkaLogConfig defaultConfig

kUtils

protected final Map<KafkaLogConfig,KafkaUtils> kUtils

Constructor Details

KafkaLogManager

public KafkaLogManager(String prefix,
 Properties producerProperties,
 Properties consumerProperties)

Since:

10.2

KafkaLogManager

public KafkaLogManager(List<KafkaLogConfig> kafkaConfigs)

Since:

11.1

Method Details

findDefaultConfig

protected KafkaLogConfig findDefaultConfig()

getConfig

protected KafkaLogConfig getConfig(Name name)

getConfig

protected KafkaLogConfig getConfig(Name name,
 Name group)

create

public void create(Name name,
 int size)

Specified by:

create in class AbstractLogManager

getSize

protected int getSize(Name name)

Specified by:

getSize in class AbstractLogManager

exists

public boolean exists(Name name)

Description copied from interface: LogManager

Returns true if a Log with this name exists.

createAppender

public <M extends Externalizable>
CloseableLogAppender<M> createAppender(Name name,
 Codec<M> codec)

Specified by:

createAppender in class AbstractLogManager

doCreateTailer

protected <M extends Externalizable> LogTailer<M> doCreateTailer(Collection<LogPartition> partitions,
 Name group,
 Codec<M> codec)

Specified by:

doCreateTailer in class AbstractLogManager

checkValidPartition

protected void checkValidPartition(LogPartition partition)

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface LogManager

Overrides:

close in class AbstractLogManager

supportSubscribe

public boolean supportSubscribe()

Description copied from interface: LogManager

Returns true if the Log LogManager.subscribe(org.nuxeo.lib.stream.log.Name, java.util.Collection<org.nuxeo.lib.stream.log.Name>, org.nuxeo.lib.stream.log.RebalanceListener, org.nuxeo.lib.stream.codec.Codec<M>) method is supported. Now deprecated because some implementations support subscribe only on specific Log names.

Specified by:

supportSubscribe in interface LogManager

Overrides:

supportSubscribe in class AbstractLogManager

supportSubscribe

public boolean supportSubscribe(Name stream)

Description copied from interface: LogManager

Returns true if the LogManager.subscribe(org.nuxeo.lib.stream.log.Name, java.util.Collection<org.nuxeo.lib.stream.log.Name>, org.nuxeo.lib.stream.log.RebalanceListener, org.nuxeo.lib.stream.codec.Codec<M>) method is supported for the specific Log.

Specified by:

supportSubscribe in interface LogManager

Overrides:

supportSubscribe in class AbstractLogManager

doSubscribe

protected <M extends Externalizable> LogTailer<M> doSubscribe(Name group,
 Collection<Name> names,
 RebalanceListener listener,
 Codec<M> codec)

Specified by:

doSubscribe in class AbstractLogManager

getLagPerPartition

public List<LogLag> getLagPerPartition(Name name,
 Name group)

Description copied from interface: LogManager

Returns the lag between consumer group and the producers for each partition. The result list is ordered, for instance index 0 is lag for partition 0.

Specified by:

getLagPerPartition in interface LogManager

Specified by:

getLagPerPartition in class AbstractLogManager

listAllNames

public List<Name> listAllNames()

Description copied from interface: LogManager

Returns all the Log names.

toString

public String toString()

Overrides:

toString in class Object

filterDisplayedProperties

protected String filterDisplayedProperties(Properties properties)

listConsumerGroups

public List<Name> listConsumerGroups(Name name)

Description copied from interface: LogManager

List the consumer groups for a Log.

delete

public boolean delete(Name name)

Description copied from interface: LogManager

Tries to delete a Log. Returns true if successfully deleted, might not be possible depending on the implementation.

Specified by:

delete in interface LogManager

Overrides:

delete in class AbstractLogManager

deleteRecords

public void deleteRecords(Name name)

Description copied from interface: LogManager

Delete all records of a stream by moving the first offsets to end of each partition. Exposed for testing purpose, might not be implemented on all implementation.

deleteConsumers

public void deleteConsumers()

Description copied from interface: LogManager

Remove all existing consumers and their committed positions. Exposed for testing purpose, might not be implemented by all implementation.