[flink] Support bounded streaming read for Fluss source - #4022
[flink] Support bounded streaming read for Fluss source#4022naivedogger wants to merge 9 commits into
Conversation
Similar to the Kafka connector's bounded read, a streaming job can now read from a given starting position up to a given stopping position and then finish, which is useful for replaying a bounded time range of the log, backfilling and archiving. - FlinkSource reports BOUNDED when stopping offsets are supplied, and passes them through createEnumerator/restoreEnumerator. - FlinkSourceEnumerator treats a streaming read with stopping offsets as bounded: one-time partition discovery and NoMoreSplits signaling, so the job finishes once all splits reach their stopping offsets. - scan.bounded.mode is supported for log tables, the changelog of primary key tables (earliest/latest/timestamp startup mode) and the $changelog/$binlog virtual tables; the full startup mode of primary key tables and the datalake union read are rejected explicitly. - FlussSourceBuilder#setBounded(OffsetsInitializer) enables bounded streaming reads in the DataStream API.
loserwang1024
left a comment
loserwang1024
left a comment
There was a problem hiding this comment.
I have left some comment. maybe we can keep the same behavior with kafka to simplify the model.
| "Only OffsetsInitializer.latest() and OffsetsInitializer.timestamp(...) are " | ||
| + "supported as stopping offsets, but was %s.", | ||
| checkedStoppingOffsetsInitializer.getClass().getName()); | ||
| this.stoppingOffsetsInitializer = checkedStoppingOffsetsInitializer; |
There was a problem hiding this comment.
please set private boolean bounded; as true, and add comment about this is place the org.apache.fluss.flink.source.FlussSourceBuilder#setBounded and compatility. And In org.apache.fluss.flink.source.FlussSourceBuilder#setBounded, set stoppingOffsetsInitializer as OffsetsInitializer.latest().
There was a problem hiding this comment.
Thanks. I kept bounded as the legacy batch-mode flag because it is used as streaming = !bounded. Setting it in the parameterized overload would incorrectly turn bounded streaming into batch execution.
The no-argument setBounded() now sets bounded = true, Boundedness.BOUNDED, and OffsetsInitializer.latest(), with a compatibility comment added.
| public static OffsetsInitializer toStoppingOffsetsInitializer(BoundedOptions boundedOptions) { | ||
| switch (boundedOptions.boundedMode) { | ||
| case UNBOUNDED: | ||
| return null; |
There was a problem hiding this comment.
Not return null, please reference kafka:
In DynamicKafkaSourceBuilder, if boundedOptions.boundedMode is UNBOUNDED, will use the default NoStoppingOffsetsInitializer:
DynamicKafkaSourceBuilder() {
this.kafkaStreamSubscriber = null;
this.kafkaMetadataService = null;
this.deserializationSchema = null;
this.startingOffsetsInitializer = OffsetsInitializer.earliest();
this.stoppingOffsetsInitializer = new NoStoppingOffsetsInitializer();
this.boundedness = Boundedness.CONTINUOUS_UNBOUNDED;
this.props = new Properties();
}There was a problem hiding this comment.
Updated. Unbounded reads now use NoStoppingOffsetsInitializer, so the stopping initializer is always non-null. Batch reads still default to OffsetsInitializer.latest().
| * user-supplied stopping offsets. A bounded read only performs a one-time partition discovery, | ||
| * since partitions created after startup are outside the bounded range captured at startup. | ||
| */ | ||
| private final boolean bounded; |
There was a problem hiding this comment.
Maybe we can use org.apache.flink.api.connector.source.Boundedness as what kafka connector does.
There was a problem hiding this comment.
Updated. Boundedness is now propagated explicitly from the builder or table source to FlinkSource and FlinkSourceEnumerator.
| streaming ? new NoStoppingOffsetsInitializer() : OffsetsInitializer.latest(); | ||
| stoppingOffsetsInitializer != null | ||
| ? stoppingOffsetsInitializer | ||
| : (streaming |
There was a problem hiding this comment.
This can done in FlussSourceBuilder.
There was a problem hiding this comment.
Updated. The stopping initializer is now resolved before constructing the enumerator, and the enumerator requires a non-null value.
| this.splitPerAssignmentBatchSize = splitPerAssignmentBatchSize; | ||
| this.initialDiscoveryFinished = initialDiscoveryFinished; | ||
| this.unassignedSplits = new ArrayList<>(unassignedSplits); | ||
| this.noMoreNewSplits = initialDiscoveryFinished && isBoundedStreamingRead(); |
There was a problem hiding this comment.
This is incorrent, we can see what kafka says:
// This flag will be marked as true if periodically partition discovery is disabled AND the
// initializing partition discovery has finished.
private boolean noMoreNewPartitionSplits = false;There was a problem hiding this comment.
Updated. I removed the constructor initialization based on restored state. noMoreNewSplits is now set only after initialization finishes when periodic partition discovery is disabled.
| } | ||
|
|
||
| if (scanPartitionDiscoveryIntervalMs > 0) { | ||
| if (restoredBoundedPartitionSet) { |
There was a problem hiding this comment.
No need to make the thread more complicate, when restart read new partition is also no pronlem.
There was a problem hiding this comment.
Updated. I removed the bounded-restore special case. A restored bounded source now follows the normal one-time partition discovery path.
| return; | ||
| } | ||
| if (t != null) { | ||
| if (isBoundedStreamingRead()) { |
There was a problem hiding this comment.
Updated. I removed the bounded-specific partition discovery failure handling and kept the existing generic behavior.
|
Thanks @loserwang1024 for the detailed review! I’ve addressed the comments and aligned the implementation more closely with Kafka’s model. Please take another look when convenient. |
2 participants
Purpose
Linked issue: close #4021
Brief change log
Tests
API and Format
Documentation