public interface SplitAssigner
extends java.io.Closeable
Assigner implementation needs to be thread safe. Enumerator call the assigner APIs mostly from
the coordinator thread. But enumerator may call the pendingSplitCount()
from the I/O threads.
Modifier and Type | Method and Description |
---|---|
default void |
close()
Some assigners may need to perform certain actions when their corresponding enumerators are
closed
|
GetSplitResult |
getNext(java.lang.String hostname)
Request a new split from the assigner when enumerator trying to assign splits to awaiting
readers.
|
java.util.concurrent.CompletableFuture<java.lang.Void> |
isAvailable()
Enumerator can get a notification via CompletableFuture when the assigner has more splits
available later.
|
default void |
onCompletedSplits(java.util.Collection<java.lang.String> completedSplitIds)
Some assigner (like event time alignment) may rack in-progress splits to advance watermark upon
completed splits
|
void |
onDiscoveredSplits(java.util.Collection<IcebergSourceSplit> splits)
Add new splits discovered by enumerator
|
void |
onUnassignedSplits(java.util.Collection<IcebergSourceSplit> splits)
Forward addSplitsBack event (for failed reader) to assigner
|
int |
pendingSplitCount()
Return the number of pending splits that haven't been assigned yet.
|
default void |
start()
Some assigners may need to start background threads or perform other activity such as
registering as listeners to updates from other event sources e.g., watermark tracker.
|
java.util.Collection<IcebergSourceSplitState> |
state()
Get assigner state for checkpointing.
|
default void start()
default void close()
close
in interface java.lang.AutoCloseable
close
in interface java.io.Closeable
GetSplitResult getNext(@Nullable java.lang.String hostname)
If enumerator wasn't able to assign the split (e.g., reader disconnected), enumerator should
call onUnassignedSplits(java.util.Collection<org.apache.iceberg.flink.source.split.IcebergSourceSplit>)
to return the split.
void onDiscoveredSplits(java.util.Collection<IcebergSourceSplit> splits)
void onUnassignedSplits(java.util.Collection<IcebergSourceSplit> splits)
default void onCompletedSplits(java.util.Collection<java.lang.String> completedSplitIds)
java.util.Collection<IcebergSourceSplitState> state()
java.util.concurrent.CompletableFuture<java.lang.Void> isAvailable()
Assigner will return the same future if this method is called again before the previous future is completed.
The future can be completed from other thread, e.g. the coordinator thread from another thread for event time alignment.
If enumerator need to trigger action upon the future completion, it may want to run it in
the coordinator thread using SplitEnumeratorContext.runInCoordinatorThread(Runnable)
.
int pendingSplitCount()
The enumerator can poll this API to publish a metric on the number of pending splits.
The enumerator can also use this information to throttle split discovery for streaming read. If there are already many pending splits tracked by the assigner, it is undesirable to discover more splits and track them in the assigner. That will increase the memory footprint and enumerator checkpoint size.
Throttling works better together with ScanContext.maxPlanningSnapshotCount()
.
Otherwise, the next split discovery after throttling will just discover all non-enumerated
snapshots and splits, which defeats the purpose of throttling.