Package org.nuxeo.lib.stream.log

Class UnifiedLogManager

java.lang.Object

org.nuxeo.lib.stream.log.UnifiedLogManager

All Implemented Interfaces:

AutoCloseable, LogManager

public class UnifiedLogManager
extends Object
implements LogManager

Since:

11.1

Field Summary

Fields

Modifier and Type	Field	Description
protected List<LogConfig>	configs
protected LogManager	cqManager
protected LogConfig	defaultConfig
protected LogManager	defaultManager
protected LogManager	kafkaManager
protected Map<LogConfig,LogManager>	managers

Constructor Summary

Constructors

Constructor	Description
UnifiedLogManager(List<LogConfig> configs)

Method Summary

Modifier and Type	Method	Description
void	close()
protected void	createCQLogManager()
boolean	createIfNotExists(Name name, int size)	Creates a new Log with size partitions if the Log does not exists.
protected void	createKafkaLogManager()
<M extends Externalizable> LogTailer<M>	createTailer(Name group, Collection<LogPartition> partitions, Codec<M> codec)	Creates a tailer for a consumer group and assign multiple partitions.
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.
boolean	exists(Name name)	Returns true if a Log with this name exists.
protected void	findDefaultLogManger()
<M extends Externalizable> LogAppender<M>	getAppender(Name name, Codec<M> codec)	Gets an appender for the Log named name, uses codec to encode records.
List<LogLag>	getLagPerPartition(Name name, Name group)	Returns the lag between consumer group and the producers for each partition.
<M extends Externalizable> List<Latency>	getLatencyPerPartition(Name name, Name group, Codec<M> codec, Function<M,Long> timestampExtractor, Function<M,String> keyExtractor)	Returns the lag with latency.
protected LogManager	getManager(Name name)
protected LogManager	getManager(Name name, Name group)
List<Name>	listAllNames()	Returns all the Log names.
List<Name>	listConsumerGroups(Name name)	List the consumer groups for a Log.
int	size(Name name)	Returns the number of partition of a Log.
<M extends Externalizable> LogTailer<M>	subscribe(Name group, Collection<Name> names, RebalanceListener listener, Codec<M> codec)	Creates a tailer for a consumer group and subscribe to multiple Logs.
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.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, 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 Detail

configs

protected final List<LogConfig> configs

cqManager

protected LogManager cqManager

kafkaManager

protected LogManager kafkaManager

defaultManager

protected LogManager defaultManager

defaultConfig

protected LogConfig defaultConfig

managers

protected Map<LogConfig,LogManager> managers

Constructor Detail

UnifiedLogManager

public UnifiedLogManager(List<LogConfig> configs)

Method Detail

createCQLogManager

protected void createCQLogManager()

createKafkaLogManager

protected void createKafkaLogManager()

findDefaultLogManger

protected void findDefaultLogManger()

getManager

protected LogManager getManager(Name name)

getManager

protected LogManager getManager(Name name,
                                Name group)

exists

public boolean exists(Name name)

Description copied from interface: LogManager

Returns true if a Log with this name exists.

Specified by:

exists in interface LogManager

createIfNotExists

public boolean createIfNotExists(Name name,
                                 int size)

Description copied from interface: LogManager

Creates a new Log with size partitions if the Log does not exists. Returns true it the Log has been created.

Specified by:

createIfNotExists in interface LogManager

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

size

public int size(Name name)

Description copied from interface: LogManager

Returns the number of partition of a Log.

Specified by:

size in interface LogManager

getAppender

public <M extends Externalizable> LogAppender<M> getAppender(Name name,
                                                             Codec<M> codec)

Description copied from interface: LogManager

Gets an appender for the Log named name, uses codec to encode records. An appender is thread safe.

Specified by:

getAppender in interface LogManager

createTailer

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

Description copied from interface: LogManager

Creates a tailer for a consumer group and assign multiple partitions. Uses codec to decode records. Note that partitions can be from different Logs. A tailer is NOT thread safe.

Specified by:

createTailer in interface LogManager

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

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

subscribe

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

Description copied from interface: LogManager

Creates a tailer for a consumer group and subscribe to multiple Logs. The partitions assignment is done dynamically depending on the number of subscribers. The partitions can change during tailers life, this is called a rebalancing. A listener can be used to be notified on assignment changes. Uses codec to decode records.

Specified by:

subscribe in interface LogManager

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

getLatencyPerPartition

public <M extends Externalizable> List<Latency> getLatencyPerPartition(Name name,
                                                                       Name group,
                                                                       Codec<M> codec,
                                                                       Function<M,Long> timestampExtractor,
                                                                       Function<M,String> keyExtractor)

Description copied from interface: LogManager

Returns the lag with latency. Timestamps used to compute the latencies are extracted from the records. This requires to read one record per partition so it costs more than LogManager.getLagPerPartition(Name, Name).

Two functions need to be provided to extract the timestamp and a key from a record.

Specified by:

getLatencyPerPartition in interface LogManager

listAllNames

public List<Name> listAllNames()

Description copied from interface: LogManager

Returns all the Log names.

Specified by:

listAllNames in interface LogManager

listConsumerGroups

public List<Name> listConsumerGroups(Name name)

Description copied from interface: LogManager

List the consumer groups for a Log.

Specified by:

listConsumerGroups in interface LogManager

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface LogManager

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.

Specified by:

deleteRecords in interface LogManager

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.

Specified by:

deleteConsumers in interface LogManager