pvary commented on code in PR #6299:
URL: https://github.com/apache/iceberg/pull/6299#discussion_r1037864725
##########
flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousIcebergEnumerator.java:
##########
@@ -92,34 +103,52 @@ public IcebergEnumeratorState snapshotState(long
checkpointId) {
/** This method is executed in an IO thread pool. */
private ContinuousEnumerationResult discoverSplits() {
- return splitPlanner.planSplits(enumeratorPosition.get());
+ int pendingSplitCountFromAssigner = assigner.pendingSplitCount();
+ if
(enumerationHistory.shouldPauseSplitDiscovery(pendingSplitCountFromAssigner)) {
+ // If the assigner already has many pending splits, it is better to
pause split discovery.
+ // Otherwise, eagerly discovering more splits will just increase
assigner memory footprint
+ // and enumerator checkpoint state size.
+ LOG.info(
+ "Pause split discovery as the assigner already has too many pending
splits: {}",
+ pendingSplitCountFromAssigner);
+ return new ContinuousEnumerationResult(
+ Collections.emptyList(), enumeratorPosition.get(),
enumeratorPosition.get());
+ } else {
+ return splitPlanner.planSplits(enumeratorPosition.get());
+ }
}
/** This method is executed in a single coordinator thread. */
private void processDiscoveredSplits(ContinuousEnumerationResult result,
Throwable error) {
if (error == null) {
if (!Objects.equals(result.fromPosition(), enumeratorPosition.get())) {
// Multiple discoverSplits() may be triggered with the same starting
snapshot to the I/O
- // thread pool.
- // E.g., the splitDiscoveryInterval is very short (like 10 ms in some
unit tests) or the
- // thread
- // pool is busy and multiple discovery actions are executed
concurrently. Discovery result
- // should
- // only be accepted if the starting position matches the enumerator
position (like
- // compare-and-swap).
+ // thread pool. E.g., the splitDiscoveryInterval is very short (like
10 ms in some unit
+ // tests) or the thread pool is busy and multiple discovery actions
are executed
+ // concurrently. Discovery result should only be accepted if the
starting position
+ // matches the enumerator position (like compare-and-swap).
LOG.info(
"Skip {} discovered splits because the scan starting position
doesn't match "
+ "the current enumerator position: enumerator position = {},
scan starting position = {}",
result.splits().size(),
enumeratorPosition.get(),
result.fromPosition());
} else {
- assigner.onDiscoveredSplits(result.splits());
- LOG.info(
- "Added {} splits discovered between ({}, {}] to the assigner",
- result.splits().size(),
- result.fromPosition(),
- result.toPosition());
+ // Sometimes, enumeration may yield no splits for a few reasons.
+ // - upstream paused or delayed streaming writes to the Iceberg table.
+ // - enumeration frequency is higher than the upstream write frequency.
+ if (!result.splits().isEmpty()) {
+ assigner.onDiscoveredSplits(result.splits());
+ // EnumerationHistory makes throttling decision on split discovery
+ // based on the total number of splits discovered in the last a few
cycles.
+ // Only update enumeration history when there are some discovered
splits.
+ enumerationHistory.add(result.splits().size());
Review Comment:
Thanks for the explanation!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
