Package org.nuxeo.lib.stream.log.internals

Class AbstractLogManager

java.lang.Object

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

All Implemented Interfaces:

AutoCloseable, LogManager

Direct Known Subclasses:

KafkaLogManager, MemLogManager

public abstract class AbstractLogManager
extends Object
implements LogManager

Field Summary

Fields

Modifier and Type	Field	Description
protected static final Name	ADMIN_GROUP
protected final Map<Name,CloseableLogAppender>	appenders
protected final Set<LogTailer>	tailers
protected final Map<LogPartitionGroup,LogTailer>	tailersAssignments

Constructor Summary

Constructors

Constructor	Description
AbstractLogManager()

Method Summary

Modifier and Type	Method	Description
protected void	checkInvalidAssignment(Name group, LogPartition partition)
protected <M extends Externalizable> void	checkInvalidCodec(LogPartition partition, Codec<M> codec)
protected void	cleanTailers()
void	close()
protected abstract void	create(Name name, int size)
protected abstract <M extends Externalizable> CloseableLogAppender<M>	createAppender(Name name, Codec<M> codec)
boolean	createIfNotExists(Name name, int size)	Creates a new Log with size partitions if the Log does not exists.
<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.
protected abstract <M extends Externalizable> LogTailer<M>	doCreateTailer(Collection<LogPartition> partitions, Name group, Codec<M> codec)
protected abstract <M extends Externalizable> LogTailer<M>	doSubscribe(Name group, Collection<Name> names, RebalanceListener listener, Codec<M> codec)
<M extends Externalizable> LogAppender<M>	getAppender(Name name, Codec<M> codec)	Gets an appender for the Log named name, uses codec to encode records.
abstract 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 abstract int	getSize(Name name)
protected <M extends Externalizable> Codec<M>	guessCodec(Collection<LogPartition> partitions)
protected <M extends Externalizable> boolean	sameCodec(Codec<M> codec1, Codec<M> codec2)
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, deleteConsumers, deleteRecords, exists, exists, getAppender, getAppender, getAppender, getLag, getLag, getLagPerPartition, getLatency, getLatency, getLatencyPerPartition, listAll, listAllNames, listConsumerGroups, listConsumerGroups, size, subscribe, subscribe, subscribe

Field Details

ADMIN_GROUP

protected static final Name ADMIN_GROUP

appenders

protected final Map<Name,CloseableLogAppender> appenders

tailersAssignments

protected final Map<LogPartitionGroup,LogTailer> tailersAssignments

tailers

protected final Set<LogTailer> tailers

Constructor Details

AbstractLogManager

public AbstractLogManager()

Method Details

create

protected abstract void create(Name name,
 int size)

getSize

protected abstract int getSize(Name name)

createAppender

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

doCreateTailer

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

doSubscribe

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

getLagPerPartition

public abstract 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

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

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

guessCodec

protected <M extends Externalizable> Codec<M> guessCodec(Collection<LogPartition> partitions)

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

checkInvalidAssignment

protected void checkInvalidAssignment(Name group,
 LogPartition partition)

checkInvalidCodec

protected <M extends Externalizable> void checkInvalidCodec(LogPartition partition,
 Codec<M> codec)

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

sameCodec

protected <M extends Externalizable> boolean sameCodec(Codec<M> codec1,
 Codec<M> codec2)

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

cleanTailers

protected void cleanTailers()

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface LogManager