Package io.confluent.parallelconsumer.state

Class ProcessingShard<K,V>

java.lang.Object

io.confluent.parallelconsumer.state.ProcessingShard<K,V>

public class ProcessingShard<K,V>
extends Object

Models the queue of work to be processed, based on the ParallelConsumerOptions.ProcessingOrder modes.

Constructor Summary

Constructors

Constructor	Description
ProcessingShard(ShardKey key, ParallelConsumerOptions<?,?> options, PartitionStateManager<K,V> pm)

Method Summary

Modifier and Type	Method	Description
void	addWorkContainer(WorkContainer<K,V> wc)
long	getCountOfWorkAwaitingSelection()
long	getCountOfWorkTracked()
long	getCountWorkInFlight()
NavigableMap<Long,WorkContainer<K,V>>	getEntries()	Map of offset to WorkUnits.
boolean	isEmpty()
void	onSuccess(WorkContainer<?,?> wc)
WorkContainer<K,V>	remove(long offset)
boolean	workIsWaitingToBeProcessed()

Methods inherited from class java.lang.Object

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

Constructor Details

ProcessingShard

public ProcessingShard(ShardKey key,
 ParallelConsumerOptions<?,?> options,
 PartitionStateManager<K,V> pm)

Method Details

workIsWaitingToBeProcessed

public boolean workIsWaitingToBeProcessed()

addWorkContainer

public void addWorkContainer(WorkContainer<K,V> wc)

onSuccess

public void onSuccess(WorkContainer<?,?> wc)

isEmpty

public boolean isEmpty()

getCountOfWorkAwaitingSelection

public long getCountOfWorkAwaitingSelection()

getCountOfWorkTracked

public long getCountOfWorkTracked()

getCountWorkInFlight

public long getCountWorkInFlight()

remove

public WorkContainer<K,V> remove(long offset)

getEntries

public NavigableMap<Long,WorkContainer<K,V>> getEntries()

Map of offset to WorkUnits.

Uses a ConcurrentSkipListMap instead of a TreeMap as under high pressure there appears to be some concurrency errors (missing WorkContainers). This is addressed in PR#270.

Is a Map because need random access into collection, as records don't always complete in order (i.e. UNORDERED mode).