You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/08/07 18:27:58 UTC
[GitHub] [flink] smattheis opened a new pull request, #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
smattheis opened a new pull request, #20485:
URL: https://github.com/apache/flink/pull/20485
## What is the purpose of the change
* The change implements FLIP-217 (https://cwiki.apache.org/confluence/display/FLINK/FLIP-217+Support+watermark+alignment+of+source+splits) to support watermark alignment of source splits.
## Brief change log
* Revised `SplitFetcher` threading model and add support to pause/resume `SplitFetcher`
* Adds support for watermark alignment of individual source splits controlled by `SourceOperator`
* Adds support for split watermark alignment for Kafka and Pulsar sources
* Adds configuration parameter to allow unaligned splits as migration plan to support legacy sources that lack support for split alignment
## Verifying this change
This change added and extended tests and can be verified as follows:
* `SplitFetcherTest` extended to test pause/resume of`SplitFetcher`
* `SourceOperatorSplitWatermarkAlignmentTest` added to test split alignment based on split reader watermarks
* `SplitFetcherPauseResumeSplitReaderITCase` added to test pause/resume of individual split readers
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): **no**
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **yes** (according to FLIP-217)
- The serializers: **no**
- The runtime per-record code paths (performance sensitive): **no**
- Anything that affects deployment or recovery: **no**
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **yes**
- If yes, how is the feature documented? **docs** (configuration parameter) and **JavaDocs** (changes of public APIs)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1246726404
Thank you everyone!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1243492777
There are still some related test failures:
> Documented option pipeline.watermark-alignment.allow-unaligned-source-splits does not exist.
and
> Sep 09 15:10:03 [ERROR] Errors:
> Sep 09 15:10:03 [ERROR] PulsarUnorderedSourceReaderTest.supportsPausingOrResumingSplits(PulsarSourceReaderBase, Boundedness, String)[1] ยป Timeout
Sep 09 15:10:03 [ERROR] PulsarUnorderedSourceReaderTest.supportsPausingOrResumingSplits(PulsarSourceReaderBase, Boundedness, String)[2] ยป Timeout
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1234480483
It looks like I will have to open a new PR because @smattheis hasn't checked the option "Allow edits by maintainers". I can't push to the branch.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1241965210
(Rebased)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r965853207
##########
flink-core/src/main/java/org/apache/flink/api/common/eventtime/CombinedWatermarkStatus.java:
##########
@@ -95,10 +97,12 @@ public boolean updateCombinedWatermark() {
static class PartialWatermark {
private long watermark = Long.MIN_VALUE;
private boolean idle = false;
+
+ @Nullable
Review Comment:
If this is only used for the sake of tests, I would actually prefer to leave this field as non-null with no-op lambda functions. This would make for easier to understand/read production code.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1245274698
Fair point, users may modify the job to disable alignment. It doesn't make sense to override the watermark alignment settings from the config. But the same argument applies to the other configuration option to disable split alignment for not-ready sources. I think it would make sense to provide a transitory option to disable split alignment altogether until this feature has proven to be stable.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r964907964
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -562,11 +580,62 @@ private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
Review Comment:
This logic is broken because `SplitWatermarkUpdateListener#updateCurrentSplitWatermark` is only called when the overall watermark of the output advances. It is not called when a split advances its watermark but the watermark is less or equal than the output's current watermark.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
+ Collection<String> splitsToPause = new ArrayList<>();
+ Collection<String> splitsToResume = new ArrayList<>();
+ splitCurrentWatermarks.forEach(
+ (splitId, splitWatermark) -> {
+ if (splitWatermark > currentMaxDesiredWatermark) {
+ splitsToPause.add(splitId);
+ } else if (currentlyPausedSplits.contains(splitId)) {
+ splitsToResume.add(splitId);
+ }
+ });
+ splitsToPause.removeAll(currentlyPausedSplits);
+ if (!splitsToPause.isEmpty() || !splitsToResume.isEmpty()) {
+ pauseOrResumeSplits(splitsToPause, splitsToResume);
+ currentlyPausedSplits.addAll(splitsToPause);
+ splitsToResume.forEach(currentlyPausedSplits::remove);
+ }
Review Comment:
Yes, I had the wrong mental model when I wrote this. The source pausing is actually independent of the split alignment, even though both mechanism use the latest max allowed watermark.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mas-chen commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mas-chen commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r944686033
##########
docs/layouts/shortcodes/generated/pipeline_configuration.html:
##########
@@ -128,5 +128,11 @@
<td>Boolean</td>
<td>Whether name of vertex includes topological index or not. When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default</td>
</tr>
+ <tr>
+ <td><h5>pipeline.watermark-alignment.allow-unaligned-source-splits</h5></td>
Review Comment:
I think `allow-aligned-source-splits` is more intuitive. The emphasis is on alignment, rather than unalignment like in checkpoints.
##########
flink-connectors/flink-connector-base/pom.xml:
##########
@@ -17,8 +17,8 @@
~ limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
Review Comment:
nit: revert unintended change?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
Review Comment:
Should this instead call `checkSplitWatermarkAlignment` to capture the case where there is only 1 split?
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherPauseResumeSplitReaderTest.java:
##########
@@ -0,0 +1,311 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.base.source.reader.fetcher;
+
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.mocks.MockSourceReader;
+import org.apache.flink.connector.base.source.reader.mocks.MockSplitReader;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.core.testutils.CommonTestUtils;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests {@link SplitFetcher} integration to pause or resume {@link SplitReader} based on {@link
+ * SourceReader} output.
+ */
+public class SplitFetcherPauseResumeSplitReaderTest {
+
+ /**
+ * Tests if pause or resume shows expected behavior which requires creation and execution of
+ * {@link SplitFetcher} tasks.
+ *
+ * @throws Exception on error.
+ */
+ @ParameterizedTest(name = "Individual reader per split: {0}")
+ @ValueSource(booleans = {false, true})
+ public void testPauseResumeSplitReaders(boolean individualReader) throws Exception {
+ final AtomicInteger numSplitReaders = new AtomicInteger();
+ final MockSplitReader.Builder readerBuilder =
+ SteppingSourceReaderTestHarness.createSplitReaderBuilder();
+ final SteppingSourceReaderTestHarness testHarness =
+ new SteppingSourceReaderTestHarness(
+ () -> {
+ numSplitReaders.getAndIncrement();
+ return readerBuilder.build();
+ },
+ new Configuration());
+
+ if (individualReader) {
+ testHarness.addPrefilledSplitsIndividualReader(2, 5);
+ assertThat(numSplitReaders.get()).isEqualTo(2);
+ } else {
+ testHarness.addPrefilledSplitsSingleReader(2, 5);
+ assertThat(numSplitReaders.get()).isEqualTo(1);
+ }
+
+ TestingReaderOutput output = new TestingReaderOutput<>();
+ testHarness.runUntilRecordsEmitted(output, 10, 2);
+ Set<Integer> recordSet = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet).containsExactlyInAnyOrder(new Integer[] {0, 1});
+
+ testHarness.pauseOrResumeSplits(Collections.singleton("0"), Collections.emptyList());
+
+ testHarness.runUntilRecordsEmitted(output, 10, 5);
+ Set<Integer> recordSet2 = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet2).containsExactlyInAnyOrder(new Integer[] {0, 1, 3, 5, 7});
+
+ testHarness.pauseOrResumeSplits(Collections.emptyList(), Collections.singleton("0"));
+
+ testHarness.runUntilAllRecordsEmitted(output, 10);
+ Set<Integer> recordSet3 = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet3)
+ .containsExactlyInAnyOrder(new Integer[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
+ }
+
+ /**
+ * Tests if pause or resume shows expected behavior in case of {@link SplitReader} that does not
+ * support split or resume for scenarios (1) allowed and (2) not allowed.
+ *
+ * @throws Exception on error.
+ */
+ @ParameterizedTest(name = "Allow unaligned source splits: {0}")
+ @ValueSource(booleans = {true, false})
+ public void testPauseResumeUnsupported(boolean allowUnalignedSourceSplits) throws Exception {
+ final AtomicInteger numSplitReaders = new AtomicInteger();
+ final Configuration configuration = new Configuration();
+ configuration.setBoolean(
+ "pipeline.watermark-alignment.allow-unaligned-source-splits",
+ allowUnalignedSourceSplits);
+ final MockSplitReader.Builder readerBuilder =
+ SteppingSourceReaderTestHarness.createSplitReaderBuilder();
+
+ final SteppingSourceReaderTestHarness testHarness =
+ new SteppingSourceReaderTestHarness(
+ () -> {
+ if (numSplitReaders.getAndIncrement() == 0) {
+ return MockSplitReaderUnsupportedPause.cloneBuilder(readerBuilder)
+ .build();
+ } else {
+ return readerBuilder.build();
+ }
+ },
+ configuration);
+
+ testHarness.addPrefilledSplitsIndividualReader(2, 5);
+ assertThat(numSplitReaders.get()).isEqualTo(2);
+
+ TestingReaderOutput output = new TestingReaderOutput<>();
+ testHarness.runUntilRecordsEmitted(output, 10, 2);
+ Set<Integer> recordSet = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet).containsExactlyInAnyOrder(new Integer[] {0, 1});
+
+ testHarness.pauseOrResumeSplits(Collections.singleton("1"), Collections.emptyList());
+
+ testHarness.runUntilRecordsEmitted(output, 10, 5);
+ Set<Integer> recordSet2 = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet2).containsExactlyInAnyOrder(new Integer[] {0, 1, 2, 4, 6});
+
+ testHarness.pauseOrResumeSplits(Collections.singleton("0"), Collections.singleton("1"));
+
+ if (allowUnalignedSourceSplits) {
+ testHarness.runUntilAllRecordsEmitted(output, 10);
+ Set<Integer> recordSet3 = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet3)
+ .containsExactlyInAnyOrder(new Integer[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
+ } else {
+ assertThatThrownBy(() -> testHarness.runUntilAllRecordsEmitted(output, 10))
+ .isInstanceOf(RuntimeException.class)
+ .hasCauseInstanceOf(UnsupportedOperationException.class);
+ }
+ }
+
+ private static class MockSteppingSplitFetcherManager<E, SplitT extends SourceSplit>
+ extends SingleThreadFetcherManager<E, SplitT> {
+
+ public MockSteppingSplitFetcherManager(
+ FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
+ Supplier<SplitReader<E, SplitT>> splitReaderSupplier,
+ Configuration configuration) {
+ super(elementsQueue, splitReaderSupplier, configuration);
+ }
+
+ @Override
+ public void addSplits(List<SplitT> splitsToAdd) {
+ SplitFetcher<E, SplitT> fetcher = createSplitFetcher();
+ fetcher.addSplits(splitsToAdd);
+ }
+
+ public void runEachOnce() {
+ for (SplitFetcher<E, SplitT> fetcher : fetchers.values()) {
+ fetcher.runOnce();
+ }
+ }
+ }
+
+ private static class MockSplitReaderUnsupportedPause extends MockSplitReader {
+ public MockSplitReaderUnsupportedPause(
+ int numRecordsPerSplitPerFetch,
+ boolean separatedFinishedRecord,
+ boolean blockingFetch) {
+ super(numRecordsPerSplitPerFetch, separatedFinishedRecord, blockingFetch);
+ }
+
+ @Override
+ public void pauseOrResumeSplits(
+ Collection<MockSourceSplit> splitsToPause,
+ Collection<MockSourceSplit> splitsToResume) {
+ throw new UnsupportedOperationException();
+ }
+
+ public static class Builder extends MockSplitReader.Builder {
+ public Builder(MockSplitReader.Builder other) {
+ super(other);
+ }
+
+ @Override
+ public MockSplitReader build() {
+ return new MockSplitReaderUnsupportedPause(
+ numRecordsPerSplitPerFetch, separatedFinishedRecord, blockingFetch);
+ }
+ }
+
+ public static Builder cloneBuilder(MockSplitReader.Builder other) {
+ return new MockSplitReaderUnsupportedPause.Builder(other);
+ }
+ }
+
+ private static class SteppingSourceReaderTestHarness {
+ private final MockSteppingSplitFetcherManager<int[], MockSourceSplit> fetcherManager;
+ private final MockSourceReader sourceReader;
+
+ public SteppingSourceReaderTestHarness(
+ Supplier<SplitReader<int[], MockSourceSplit>> splitReaderSupplier,
+ Configuration configuration) {
+ FutureCompletingBlockingQueue<RecordsWithSplitIds<int[]>> queue =
+ new FutureCompletingBlockingQueue<>(10);
+ this.fetcherManager =
+ new MockSteppingSplitFetcherManager<>(
+ queue, splitReaderSupplier, configuration);
+ this.sourceReader =
+ new MockSourceReader(
+ queue, fetcherManager, configuration, new TestingReaderContext());
+ }
+
+ private static List<MockSourceSplit> createPrefilledSplits(int numSplits, int numRecords) {
+ final List<MockSourceSplit> splits = new ArrayList<>(numSplits);
+ for (int splitId = 0; splitId < numSplits; splitId++) {
+ MockSourceSplit split = new MockSourceSplit(splitId, 0, numRecords);
+ for (int i = 0; i < numRecords; i++) {
+ split.addRecord(i * numSplits + splitId);
+ }
+ splits.add(split);
+ }
+ return splits;
+ }
+
+ public void addPrefilledSplitsSingleReader(int numSplits, int numRecords) {
+ sourceReader.addSplits(createPrefilledSplits(numSplits, numRecords));
+ sourceReader.notifyNoMoreSplits();
Review Comment:
Is this a required assumption? We should also exercise the case where splits are dynamically discovered (e.g. Kafka source topic partition discovery).
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/PauseOrResumeSplitsTask.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.base.source.reader.fetcher;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collection;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Changes the paused splits of a n{@link SplitReader}. The task is used by default in {@link
+ * SplitFetcherManager} and assumes that a {@link SplitFetcher} has multiple splits. For {@code
+ * SplitFetchers} with single splits, it's instead recommended to subclass {@link
+ * SplitFetcherManager} and pause the whole {@code SplitFetcher}.
+ *
+ * @param <SplitT> the type of the split
+ */
+@Internal
+class PauseOrResumeSplitsTask<SplitT extends SourceSplit> implements SplitFetcherTask {
+ private static final Logger LOG = LoggerFactory.getLogger(PauseOrResumeSplitsTask.class);
+ private final SplitReader<?, SplitT> splitReader;
+ private final Collection<SplitT> splitsToPause;
+ private final Collection<SplitT> splitsToResume;
+ private final boolean allowUnalignedSourceSplits;
+
+ PauseOrResumeSplitsTask(
+ SplitReader<?, SplitT> splitReader,
+ Collection<SplitT> splitsToPause,
+ Collection<SplitT> splitsToResume,
+ boolean allowUnalignedSourceSplits) {
+ this.splitReader = checkNotNull(splitReader);
+ this.splitsToPause = checkNotNull(splitsToPause);
+ this.splitsToResume = checkNotNull(splitsToResume);
+ this.allowUnalignedSourceSplits = allowUnalignedSourceSplits;
+ }
+
+ @Override
+ public boolean run() throws IOException {
+ try {
+ splitReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ } catch (UnsupportedOperationException e) {
+ if (!allowUnalignedSourceSplits) {
+ throw new UnsupportedOperationException("", e);
Review Comment:
Can we rethrow exception without unnecessary wrapping? `throw e`
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java:
##########
@@ -0,0 +1,144 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.eventtime.Watermark;
+import org.apache.flink.api.common.eventtime.WatermarkGenerator;
+import org.apache.flink.api.common.eventtime.WatermarkOutput;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.connector.source.mocks.MockSourceReader;
+import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
+import org.apache.flink.api.connector.source.mocks.MockSourceSplitSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.operators.coordination.MockOperatorEventGateway;
+import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
+import org.apache.flink.runtime.source.event.AddSplitEvent;
+import org.apache.flink.runtime.source.event.WatermarkAlignmentEvent;
+import org.apache.flink.runtime.state.TestTaskStateManager;
+import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.streaming.api.operators.source.CollectingDataOutput;
+import org.apache.flink.streaming.api.operators.source.TestingSourceOperator;
+import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
+import org.apache.flink.streaming.util.MockOutput;
+import org.apache.flink.streaming.util.MockStreamConfig;
+
+import org.junit.Test;
Review Comment:
nit: use junit5
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReader.java:
##########
@@ -132,4 +139,28 @@ default void handleSourceEvents(SourceEvent sourceEvent) {}
*/
@Override
default void notifyCheckpointComplete(long checkpointId) throws Exception {}
+
+ /**
+ * Pauses or resumes reading of individual source splits.
+ *
+ * <p>Note that no other methods can be called in parallel, so it's fine to non-atomically
Review Comment:
nit: This reads unclearly. What about: ", so updating subscriptions can be atomically updated".
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
+ Collection<String> splitsToPause = new ArrayList<>();
+ Collection<String> splitsToResume = new ArrayList<>();
+ splitCurrentWatermarks.forEach(
+ (splitId, splitWatermark) -> {
+ if (splitWatermark > currentMaxDesiredWatermark) {
+ splitsToPause.add(splitId);
+ } else if (currentlyPausedSplits.contains(splitId)) {
+ splitsToResume.add(splitId);
+ }
+ });
+ splitsToPause.removeAll(currentlyPausedSplits);
+ if (!splitsToPause.isEmpty() || !splitsToResume.isEmpty()) {
+ pauseOrResumeSplits(splitsToPause, splitsToResume);
+ currentlyPausedSplits.addAll(splitsToPause);
+ splitsToResume.forEach(currentlyPausedSplits::remove);
+ }
+ }
+
+ private void pauseOrResumeSplits(
+ Collection<String> splitsToPause, Collection<String> splitsToResume) {
+ if (!allowUnalignedSourceSplits || sourceReaderSupportsPauseOrResumeSplits) {
+ try {
+ sourceReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ } catch (UnsupportedOperationException e) {
+ sourceReaderSupportsPauseOrResumeSplits = false;
+ if (!allowUnalignedSourceSplits) {
+ throw new UnsupportedOperationException("", e);
Review Comment:
ditto on wrapping exception
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r965653484
##########
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java:
##########
@@ -179,6 +180,21 @@ public void handleSplitsChanges(SplitsChange<PulsarPartitionSplit> splitsChanges
this.pulsarConsumer = consumer;
}
+ @Override
+ public void pauseOrResumeSplits(
+ Collection<PulsarPartitionSplit> splitsToPause,
+ Collection<PulsarPartitionSplit> splitsToResume) {
+ if (splitsToPause.size() > 1 || splitsToResume.size() > 1) {
+ throw new IllegalStateException("This pulsar split reader only support one split.");
Review Comment:
Have you found an answer for that @mxm ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1240464909
> The next question would be how to squash the commits? If you want to preserve authorship of your changes @mxm, without overwriting authors of the base commits, we would need either a single fixup commit from you, all one your fixup per original commit.
I'd prefer this to be a single suqashed commit with the use of Co-Authored-By. IMHO there is no benefit in merging the individual commits because they are very much tight to each other. This keeps the history clean while retaining authorship. What do you think?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r964978381
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -562,11 +580,62 @@ private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
Review Comment:
Actually, that's ok. It simply means that the split watermark of a split may not be registered if its watermark is always lagging behind another splits watermark. Either this split will overtake the other split and become registered here, or the other split will only ever get paused.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] tweise commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
tweise commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1246779270
Exciting to see this make the finish line! This was a long awaited feature (we implemented alignment for the Kinesis source ~4 years ago). Let's see how it holds up in production use cases!
Thanks @mxm @pnowojski @smattheis @dawidwys, @AHeise for your contributions!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r961619833
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -282,4 +282,21 @@ public enum VertexDescriptionMode {
.withDescription(
"Whether name of vertex includes topological index or not. "
+ "When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default");
+
+ @PublicEvolving
+ public static final ConfigOption<Boolean> ALLOW_UNALIGNED_SOURCE_SPLITS =
+ key("pipeline.watermark-alignment.allow-unaligned-source-splits")
+ .booleanType()
+ .defaultValue(false)
Review Comment:
> watermark alignment without split alignment still works but it is possible for splits to overtake each other which can slow down the processing due to regular watermark alignment
I'm not sure if I understand you here. So let me rephrase the consensus.
On the current master. Watermark alignment works to a small extent if a source operator has more than one split, but it doesn't work fully. In worst case, it won't work at all. Imagine if the slowest split, that is a split that's the most lagging/blocking watermark emission, is processed on the same operator that has the fastest split. In such scenario, watermark alignment without per split alignment, can not block processing of this whole operator (or otherwise it would deadlock the watermarks emission), so the fastest split will continue producing records, for example blowing up time windowed stated downstream. Even in average case, such watermark alignment will work very badly.
Moreover, using pre-FLIP-217 watermark alignment with multiple splits per operator has been always discouraged. With FLIP-217 the consensus was, that it would be actually hurtful for the users, to have _SILENTLY_ hit such scenarios: user could configure watermark alignment, but actually it wouldn't work. The consensus was that from now on this feature should either work fully, or not at all if users tries to enable it. But we also want to provide gradual migration path for any users that would be affected by this change. Instead of forcing user to both upgrade Flink version AND solve the lack of per-split alignment on his side, he can temporarily set `allow-unaligned-source-splits` to `true`, do the Flink upgrade, and then in the background prepare for next Flink release that will remove this option and set it to `false` in stone.
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -282,4 +282,21 @@ public enum VertexDescriptionMode {
.withDescription(
"Whether name of vertex includes topological index or not. "
+ "When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default");
+
+ @PublicEvolving
+ public static final ConfigOption<Boolean> ALLOW_UNALIGNED_SOURCE_SPLITS =
Review Comment:
This should be marked as `@Deprecated("Will be removed in future Flink releases")`.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
Review Comment:
What problem do you see with this? The logic of both of those methods looks ok and consistent to me. Are you worried about threading issues? Both of those methods are called only from the task thread.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
+ Collection<String> splitsToPause = new ArrayList<>();
+ Collection<String> splitsToResume = new ArrayList<>();
+ splitCurrentWatermarks.forEach(
+ (splitId, splitWatermark) -> {
+ if (splitWatermark > currentMaxDesiredWatermark) {
+ splitsToPause.add(splitId);
+ } else if (currentlyPausedSplits.contains(splitId)) {
+ splitsToResume.add(splitId);
+ }
+ });
+ splitsToPause.removeAll(currentlyPausedSplits);
+ if (!splitsToPause.isEmpty() || !splitsToResume.isEmpty()) {
+ pauseOrResumeSplits(splitsToPause, splitsToResume);
+ currentlyPausedSplits.addAll(splitsToPause);
+ splitsToResume.forEach(currentlyPausedSplits::remove);
+ }
Review Comment:
๐ So as user you mean the flink-runtime? :)
I think should be indeed checked and asserted via a test. I would expect this to work - if all splits are paused, eventually (maybe after a one or two extra polls) the source should return no more input available.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r964820287
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
Review Comment:
I agree, this is inconsistent in the case of a single split. I would add another check here for the number of splits. Calling checkSplitWatermarkAlignment is slightly more expensive due to it checking all splits.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] smattheis commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
smattheis commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1240941359
@mxm Thanks a lot also from my side for you contributions. However, @pnowojski this PR was not orphaned! (I wrote to you that I'm on vacations.)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r966251899
##########
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java:
##########
@@ -179,6 +180,21 @@ public void handleSplitsChanges(SplitsChange<PulsarPartitionSplit> splitsChanges
this.pulsarConsumer = consumer;
}
+ @Override
+ public void pauseOrResumeSplits(
+ Collection<PulsarPartitionSplit> splitsToPause,
+ Collection<PulsarPartitionSplit> splitsToResume) {
+ if (splitsToPause.size() > 1 || splitsToResume.size() > 1) {
+ throw new IllegalStateException("This pulsar split reader only support one split.");
Review Comment:
Yes, Pulsar splits are always started in a separate fetcher thread. We still want to be able to pause these split threads. So the code makes sense.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r963029722
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -282,4 +282,21 @@ public enum VertexDescriptionMode {
.withDescription(
"Whether name of vertex includes topological index or not. "
+ "When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default");
+
+ @PublicEvolving
+ public static final ConfigOption<Boolean> ALLOW_UNALIGNED_SOURCE_SPLITS =
Review Comment:
`@Deprecated` doesn't allow for a string as argument but I've added an additional comment.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mas-chen commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mas-chen commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r946433696
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherTest.java:
##########
@@ -255,6 +256,23 @@ public void testClose() {
assertThat(splitReader.isClosed()).isTrue();
}
+ @Test
+ public void testCloseAfterPause() {
+ final FutureCompletingBlockingQueue<RecordsWithSplitIds<Object>> queue =
+ new FutureCompletingBlockingQueue<>();
+ final SplitFetcher<Object, TestingSourceSplit> fetcher =
+ createFetcherWithSplit(
+ "test-split",
+ queue,
+ new TestingSplitReader<>(finishedSplitFetch("test-split")));
+
+ fetcher.pause();
+
+ new Thread(fetcher::shutdown).start();
+
+ fetcher.runOnce();
Review Comment:
Can we assert that this is false?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r958152581
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
Review Comment:
Would you be willing to simplify this? If so, I would suggest to do it as a separate commit (either in the same PR or a new PR).
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
+ Collection<String> splitsToPause = new ArrayList<>();
+ Collection<String> splitsToResume = new ArrayList<>();
+ splitCurrentWatermarks.forEach(
+ (splitId, splitWatermark) -> {
+ if (splitWatermark > currentMaxDesiredWatermark) {
+ splitsToPause.add(splitId);
+ } else if (currentlyPausedSplits.contains(splitId)) {
+ splitsToResume.add(splitId);
+ }
+ });
+ splitsToPause.removeAll(currentlyPausedSplits);
+ if (!splitsToPause.isEmpty() || !splitsToResume.isEmpty()) {
+ pauseOrResumeSplits(splitsToPause, splitsToResume);
+ currentlyPausedSplits.addAll(splitsToPause);
+ splitsToResume.forEach(currentlyPausedSplits::remove);
+ }
+ }
+
+ private void pauseOrResumeSplits(
+ Collection<String> splitsToPause, Collection<String> splitsToResume) {
+ if (!allowUnalignedSourceSplits || sourceReaderSupportsPauseOrResumeSplits) {
+ try {
+ sourceReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ } catch (UnsupportedOperationException e) {
+ sourceReaderSupportsPauseOrResumeSplits = false;
+ if (!allowUnalignedSourceSplits) {
+ throw new UnsupportedOperationException("", e);
Review Comment:
> In my understanding, readers which do not support pausing splits will fail the job, unless the configuration to allow unaligned splits is set. Hence, I don't understand the sourceReaderSupportsPauseOrResumeSplits flag.
please check my comment above.
##########
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -227,6 +227,20 @@ public void close() throws Exception {
consumer.close();
}
+ @Override
+ public void pauseOrResumeSplits(
+ Collection<KafkaPartitionSplit> splitsToPause,
+ Collection<KafkaPartitionSplit> splitsToResume) {
+ consumer.resume(
+ splitsToResume.stream()
+ .map(KafkaPartitionSplit::getTopicPartition)
+ .collect(Collectors.toList()));
+ consumer.pause(
+ splitsToPause.stream()
+ .map(KafkaPartitionSplit::getTopicPartition)
+ .collect(Collectors.toList()));
Review Comment:
I'm not sure how likely is this `resume` to ever block and even if it blocks, I don't think it should cause any problems. Keep in mind that this is not an exact feature. We do not guarantee that no records will be processed . The whole watermark alignment requires sending RPCs between TMs and JM so it has built in delay/lag.
At the same time I don't see how doing this in the other way around would cause much harm. At worst the consumer would have a for a very short period of time no splits to process if everything is paused but not yet resumed.
##########
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java:
##########
@@ -179,6 +180,21 @@ public void handleSplitsChanges(SplitsChange<PulsarPartitionSplit> splitsChanges
this.pulsarConsumer = consumer;
}
+ @Override
+ public void pauseOrResumeSplits(
+ Collection<PulsarPartitionSplit> splitsToPause,
+ Collection<PulsarPartitionSplit> splitsToResume) {
+ if (splitsToPause.size() > 1 || splitsToResume.size() > 1) {
+ throw new IllegalStateException("This pulsar split reader only support one split.");
Review Comment:
Good question...
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
Review Comment:
? This line that you are pointing to
```
plitsToPause.removeAll(currentlyPausedSplits);
```
doesn't modify `currentlyPausedSplits`. It only gathers the information what splits have to be paused/resumed.
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcher.java:
##########
@@ -171,17 +230,54 @@ void runOnce() {
* @param splitsToAdd the splits to add.
*/
public void addSplits(List<SplitT> splitsToAdd) {
- enqueueTask(new AddSplitsTask<>(splitReader, splitsToAdd, assignedSplits));
- wakeUp(true);
+ lock.lock();
+ try {
+ enqueueTaskUnsafe(new AddSplitsTask<>(splitReader, splitsToAdd, assignedSplits));
+ wakeUpUnsafe(true);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Called when some splits of this source instance progressed too much beyond the global
+ * watermark of all subtasks. If the split reader implements {@link SplitReader}, it will relay
+ * the information asynchronously through the split fetcher thread.
+ *
+ * @param splitsToPause the splits to pause
+ * @param splitsToResume the splits to resume
+ */
+ public void pauseOrResumeSplits(
+ Collection<SplitT> splitsToPause, Collection<SplitT> splitsToResume) {
+ lock.lock();
+ try {
+ enqueueTaskUnsafe(
+ new PauseOrResumeSplitsTask<>(
+ splitReader,
+ splitsToPause,
+ splitsToResume,
+ allowUnalignedSourceSplits));
+ wakeUpUnsafe(true);
+ } finally {
+ lock.unlock();
+ }
}
public void enqueueTask(SplitFetcherTask task) {
- synchronized (lock) {
- taskQueue.offer(task);
- isIdle = false;
+ lock.lock();
+ try {
+ enqueueTaskUnsafe(task);
+ } finally {
+ lock.unlock();
}
}
+ private void enqueueTaskUnsafe(SplitFetcherTask task) {
+ assert lock.isHeldByCurrentThread();
Review Comment:
More or less this is our convention (at least in the runtime) to use those `Unsafe` suffixed methods. As far as I remember double acquisition of the same lock is extremely expensive in JVM (10x slower?).
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
+ Collection<String> splitsToPause = new ArrayList<>();
+ Collection<String> splitsToResume = new ArrayList<>();
+ splitCurrentWatermarks.forEach(
+ (splitId, splitWatermark) -> {
+ if (splitWatermark > currentMaxDesiredWatermark) {
+ splitsToPause.add(splitId);
+ } else if (currentlyPausedSplits.contains(splitId)) {
+ splitsToResume.add(splitId);
+ }
+ });
+ splitsToPause.removeAll(currentlyPausedSplits);
+ if (!splitsToPause.isEmpty() || !splitsToResume.isEmpty()) {
+ pauseOrResumeSplits(splitsToPause, splitsToResume);
+ currentlyPausedSplits.addAll(splitsToPause);
+ splitsToResume.forEach(currentlyPausedSplits::remove);
+ }
Review Comment:
What do you mean "signal to user"?
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -282,4 +282,21 @@ public enum VertexDescriptionMode {
.withDescription(
"Whether name of vertex includes topological index or not. "
+ "When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default");
+
+ @PublicEvolving
+ public static final ConfigOption<Boolean> ALLOW_UNALIGNED_SOURCE_SPLITS =
+ key("pipeline.watermark-alignment.allow-unaligned-source-splits")
+ .booleanType()
+ .defaultValue(false)
Review Comment:
Please check the dev mailing list discussion. We want to change the the old behaviour from FLINK-1.14/1.15. In a couple of releases (1.18? 1.19?), we will want to disallow partially working watermark alignment when source doesn't support per split alignment. At the same time we want to provide users time to adjust/migrate if they are relaying on this partially working feature for now.
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java:
##########
@@ -142,6 +143,30 @@ public void accept(Throwable t) {
public abstract void addSplits(List<SplitT> splitsToAdd);
+ public void alignSplitsWatermarks(
Review Comment:
nitty nit: this should be called `pauseOrResumeSplits` in the first commit that it appears.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r960476385
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
+ Collection<String> splitsToPause = new ArrayList<>();
+ Collection<String> splitsToResume = new ArrayList<>();
+ splitCurrentWatermarks.forEach(
+ (splitId, splitWatermark) -> {
+ if (splitWatermark > currentMaxDesiredWatermark) {
+ splitsToPause.add(splitId);
+ } else if (currentlyPausedSplits.contains(splitId)) {
+ splitsToResume.add(splitId);
+ }
+ });
+ splitsToPause.removeAll(currentlyPausedSplits);
+ if (!splitsToPause.isEmpty() || !splitsToResume.isEmpty()) {
+ pauseOrResumeSplits(splitsToPause, splitsToResume);
+ currentlyPausedSplits.addAll(splitsToPause);
+ splitsToResume.forEach(currentlyPausedSplits::remove);
+ }
Review Comment:
I thought we might expose DataInputStatus of the SourceOperator somewhere. Semantically, one split being paused should yield the same status as multiple splits being all paused.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r963030325
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherTest.java:
##########
@@ -255,6 +256,23 @@ public void testClose() {
assertThat(splitReader.isClosed()).isTrue();
}
+ @Test
+ public void testCloseAfterPause() {
+ final FutureCompletingBlockingQueue<RecordsWithSplitIds<Object>> queue =
+ new FutureCompletingBlockingQueue<>();
+ final SplitFetcher<Object, TestingSourceSplit> fetcher =
+ createFetcherWithSplit(
+ "test-split",
+ queue,
+ new TestingSplitReader<>(finishedSplitFetch("test-split")));
+
+ fetcher.pause();
+
+ new Thread(fetcher::shutdown).start();
+
+ fetcher.runOnce();
Review Comment:
Done.
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -282,4 +282,21 @@ public enum VertexDescriptionMode {
.withDescription(
"Whether name of vertex includes topological index or not. "
+ "When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default");
+
+ @PublicEvolving
+ public static final ConfigOption<Boolean> ALLOW_UNALIGNED_SOURCE_SPLITS =
Review Comment:
Deprecated doesn't allow for a string as argument but I've added an additional comment.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1245315606
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1240828114
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1243913279
> On another note, do you think it makes sense to add a configuration option to disable using split alignment entirely? I'm a bit worried this feature might cause issues with the already integrated Kafka/Pulsar sources that we are not foreseeing right now. The added configuration option only allows to disable split alignments for non-compatible connectors. We could add another option to disable it entirely to retain the old behavior (disabled by default).
Does it make sense to use and enable watermark alignment via `DataStream API` to only disable it via config value? Keep in mind that user has to manually enable via `WatermarkStrategy#withWatermarkAlignment` call [1].
[1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/event-time/generating_watermarks/#watermark-alignment-_beta_
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1207462269
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5c245dda9e66849ae58ed6503e37f52b59e9624e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5c245dda9e66849ae58ed6503e37f52b59e9624e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5c245dda9e66849ae58ed6503e37f52b59e9624e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1243567879
> There are still some related test failures:
>
> > Documented option pipeline.watermark-alignment.allow-unaligned-source-splits does not exist.
I had to remove the `@Deprecated` flag from the configuration option. Otherwise, it will be undocumented.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1243568641
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r952615289
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java:
##########
@@ -0,0 +1,144 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.eventtime.Watermark;
+import org.apache.flink.api.common.eventtime.WatermarkGenerator;
+import org.apache.flink.api.common.eventtime.WatermarkOutput;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.connector.source.mocks.MockSourceReader;
+import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
+import org.apache.flink.api.connector.source.mocks.MockSourceSplitSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.operators.coordination.MockOperatorEventGateway;
+import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
+import org.apache.flink.runtime.source.event.AddSplitEvent;
+import org.apache.flink.runtime.source.event.WatermarkAlignmentEvent;
+import org.apache.flink.runtime.state.TestTaskStateManager;
+import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.streaming.api.operators.source.CollectingDataOutput;
+import org.apache.flink.streaming.api.operators.source.TestingSourceOperator;
+import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
+import org.apache.flink.streaming.util.MockOutput;
+import org.apache.flink.streaming.util.MockStreamConfig;
+
+import org.junit.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit test for split alignment in {@link SourceOperator}. */
+public class SourceOperatorSplitWatermarkAlignmentTest {
+ public static final WatermarkGenerator<Integer> WATERMARK_GENERATOR =
+ new WatermarkGenerator<Integer>() {
+
+ private long maxWatermark = Long.MIN_VALUE;
+
+ @Override
+ public void onEvent(Integer event, long eventTimestamp, WatermarkOutput output) {
+ if (eventTimestamp > maxWatermark) {
+ this.maxWatermark = eventTimestamp;
+ output.emitWatermark(new Watermark(maxWatermark));
+ }
+ }
+
+ @Override
+ public void onPeriodicEmit(WatermarkOutput output) {
+ output.emitWatermark(new Watermark(maxWatermark));
+ }
+ };
+
+ @Test
+ public void testSplitWatermarkAlignment() throws Exception {
+
+ final SplitAligningSourceReader sourceReader = new SplitAligningSourceReader();
+ SourceOperator<Integer, MockSourceSplit> operator =
+ new TestingSourceOperator<>(
+ sourceReader,
+ WatermarkStrategy.forGenerator(ctx -> WATERMARK_GENERATOR)
+ .withTimestampAssigner((r, l) -> r)
+ .withWatermarkAlignment("group-1", Duration.ofMillis(1)),
+ new TestProcessingTimeService(),
+ new MockOperatorEventGateway(),
+ 1,
+ 5,
+ true);
+ Environment env = getTestingEnvironment();
+ operator.setup(
+ new SourceOperatorStreamTask<Integer>(env),
+ new MockStreamConfig(new Configuration(), 1),
+ new MockOutput<>(new ArrayList<>()));
+ operator.initializeState(new StreamTaskStateInitializerImpl(env, new MemoryStateBackend()));
+
+ operator.open();
+ final MockSourceSplit split1 = new MockSourceSplit(0, 0, 10);
+ final MockSourceSplit split2 = new MockSourceSplit(1, 10, 20);
+ split1.addRecord(5);
+ split1.addRecord(6);
+ split1.addRecord(11);
+ operator.handleOperatorEvent(
+ new AddSplitEvent<>(
+ Arrays.asList(split1, split2), new MockSourceSplitSerializer()));
+ final CollectingDataOutput<Integer> dataOutput = new CollectingDataOutput<>();
+
+ operator.emitNext(dataOutput); // 5
+ operator.handleOperatorEvent(
+ new WatermarkAlignmentEvent(4)); // pause by coordinator message
+ assertThat(sourceReader.pausedSplits).containsExactly("0");
+ operator.handleOperatorEvent(new WatermarkAlignmentEvent(5));
+ assertThat(sourceReader.pausedSplits).isEmpty();
+ operator.emitNext(dataOutput); // 6, pause by increasing watermark
+ assertThat(sourceReader.pausedSplits).containsExactly("0");
+ }
Review Comment:
Should we test resuming as well?
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/PauseOrResumeSplitsTask.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.base.source.reader.fetcher;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collection;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Changes the paused splits of a n{@link SplitReader}. The task is used by default in {@link
+ * SplitFetcherManager} and assumes that a {@link SplitFetcher} has multiple splits. For {@code
+ * SplitFetchers} with single splits, it's instead recommended to subclass {@link
+ * SplitFetcherManager} and pause the whole {@code SplitFetcher}.
+ *
+ * @param <SplitT> the type of the split
+ */
+@Internal
+class PauseOrResumeSplitsTask<SplitT extends SourceSplit> implements SplitFetcherTask {
+ private static final Logger LOG = LoggerFactory.getLogger(PauseOrResumeSplitsTask.class);
+ private final SplitReader<?, SplitT> splitReader;
+ private final Collection<SplitT> splitsToPause;
+ private final Collection<SplitT> splitsToResume;
+ private final boolean allowUnalignedSourceSplits;
+
+ PauseOrResumeSplitsTask(
+ SplitReader<?, SplitT> splitReader,
+ Collection<SplitT> splitsToPause,
+ Collection<SplitT> splitsToResume,
+ boolean allowUnalignedSourceSplits) {
+ this.splitReader = checkNotNull(splitReader);
+ this.splitsToPause = checkNotNull(splitsToPause);
+ this.splitsToResume = checkNotNull(splitsToResume);
+ this.allowUnalignedSourceSplits = allowUnalignedSourceSplits;
+ }
+
+ @Override
+ public boolean run() throws IOException {
+ try {
+ splitReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ } catch (UnsupportedOperationException e) {
+ if (!allowUnalignedSourceSplits) {
+ throw new UnsupportedOperationException("", e);
Review Comment:
+1
##########
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -227,6 +227,20 @@ public void close() throws Exception {
consumer.close();
}
+ @Override
+ public void pauseOrResumeSplits(
+ Collection<KafkaPartitionSplit> splitsToPause,
+ Collection<KafkaPartitionSplit> splitsToResume) {
+ consumer.resume(
+ splitsToResume.stream()
+ .map(KafkaPartitionSplit::getTopicPartition)
+ .collect(Collectors.toList()));
+ consumer.pause(
+ splitsToPause.stream()
+ .map(KafkaPartitionSplit::getTopicPartition)
+ .collect(Collectors.toList()));
Review Comment:
Should the order be the other way around, in case resuming blocks for any reason?
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcher.java:
##########
@@ -171,17 +230,54 @@ void runOnce() {
* @param splitsToAdd the splits to add.
*/
public void addSplits(List<SplitT> splitsToAdd) {
- enqueueTask(new AddSplitsTask<>(splitReader, splitsToAdd, assignedSplits));
- wakeUp(true);
+ lock.lock();
+ try {
+ enqueueTaskUnsafe(new AddSplitsTask<>(splitReader, splitsToAdd, assignedSplits));
+ wakeUpUnsafe(true);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Called when some splits of this source instance progressed too much beyond the global
+ * watermark of all subtasks. If the split reader implements {@link SplitReader}, it will relay
+ * the information asynchronously through the split fetcher thread.
+ *
+ * @param splitsToPause the splits to pause
+ * @param splitsToResume the splits to resume
+ */
+ public void pauseOrResumeSplits(
+ Collection<SplitT> splitsToPause, Collection<SplitT> splitsToResume) {
+ lock.lock();
+ try {
+ enqueueTaskUnsafe(
+ new PauseOrResumeSplitsTask<>(
+ splitReader,
+ splitsToPause,
+ splitsToResume,
+ allowUnalignedSourceSplits));
+ wakeUpUnsafe(true);
+ } finally {
+ lock.unlock();
+ }
}
public void enqueueTask(SplitFetcherTask task) {
- synchronized (lock) {
- taskQueue.offer(task);
- isIdle = false;
+ lock.lock();
+ try {
+ enqueueTaskUnsafe(task);
+ } finally {
+ lock.unlock();
}
}
+ private void enqueueTaskUnsafe(SplitFetcherTask task) {
+ assert lock.isHeldByCurrentThread();
Review Comment:
Since this is an ReentrantLock, we could also lock here which would make this method safer. The assert won't run in production.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
+ Collection<String> splitsToPause = new ArrayList<>();
+ Collection<String> splitsToResume = new ArrayList<>();
+ splitCurrentWatermarks.forEach(
+ (splitId, splitWatermark) -> {
+ if (splitWatermark > currentMaxDesiredWatermark) {
+ splitsToPause.add(splitId);
+ } else if (currentlyPausedSplits.contains(splitId)) {
+ splitsToResume.add(splitId);
+ }
+ });
+ splitsToPause.removeAll(currentlyPausedSplits);
+ if (!splitsToPause.isEmpty() || !splitsToResume.isEmpty()) {
+ pauseOrResumeSplits(splitsToPause, splitsToResume);
+ currentlyPausedSplits.addAll(splitsToPause);
+ splitsToResume.forEach(currentlyPausedSplits::remove);
+ }
+ }
+
+ private void pauseOrResumeSplits(
+ Collection<String> splitsToPause, Collection<String> splitsToResume) {
+ if (!allowUnalignedSourceSplits || sourceReaderSupportsPauseOrResumeSplits) {
+ try {
+ sourceReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ } catch (UnsupportedOperationException e) {
+ sourceReaderSupportsPauseOrResumeSplits = false;
+ if (!allowUnalignedSourceSplits) {
+ throw new UnsupportedOperationException("", e);
Review Comment:
+1
##########
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java:
##########
@@ -179,6 +180,21 @@ public void handleSplitsChanges(SplitsChange<PulsarPartitionSplit> splitsChanges
this.pulsarConsumer = consumer;
}
+ @Override
+ public void pauseOrResumeSplits(
+ Collection<PulsarPartitionSplit> splitsToPause,
+ Collection<PulsarPartitionSplit> splitsToResume) {
+ if (splitsToPause.size() > 1 || splitsToResume.size() > 1) {
+ throw new IllegalStateException("This pulsar split reader only support one split.");
Review Comment:
I don't understand why we can't pause single splits. Even if so, I don't know why that should leak to the Pulsar reader.
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcher.java:
##########
@@ -210,138 +313,73 @@ Map<String, SplitT> assignedSplits() {
/**
* Package private for unit test.
*
- * @return true if task queue is not empty, false otherwise.
+ * @return true if task queue is empty, false otherwise.
*/
boolean isIdle() {
- return isIdle;
- }
-
- /**
- * Check whether the fetch task should run. The fetch task should only run when all the
- * following conditions are met. 1. there is no task in the task queue to run. 2. there are
- * assigned splits Package private for testing purpose.
- *
- * @return whether the fetch task should be run.
- */
- boolean shouldRunFetchTask() {
- return taskQueue.isEmpty() && !assignedSplits.isEmpty();
+ lock.lock();
+ try {
+ return assignedSplits.isEmpty() && taskQueue.isEmpty() && runningTask == null;
+ } finally {
+ lock.unlock();
+ }
}
/**
* Wake up the fetcher thread. There are only two blocking points in a running fetcher. 1.
- * Taking the next task out of the task queue. 2. Running a task.
+ * Waiting for the next task in an idle fetcher. 2. Running a task.
*
* <p>They need to be waken up differently. If the fetcher is blocking waiting on the next task
- * in the task queue, we should just interrupt the fetcher thread. If the fetcher is running the
- * user split reader, we should call SplitReader.wakeUp() instead of naively interrupt the
- * thread.
- *
- * <p>The correctness can be think of in the following way. The purpose of wake up is to let the
- * fetcher thread go to the very beginning of the running loop. There are three major events in
- * each run of the loop.
- *
- * <ol>
- * <li>pick a task (blocking)
- * <li>assign the task to runningTask variable.
- * <li>run the runningTask. (blocking)
- * </ol>
- *
- * <p>We don't need to worry about things after step 3 because there is no blocking point
- * anymore.
- *
- * <p>We always first set the wakeup flag when waking up the fetcher, then use the value of
- * running task to determine where the fetcher thread is.
- *
- * <ul>
- * <li>If runningThread is null, it is before step 2, so we should interrupt fetcher. This
- * interruption will not be propagated to the split reader, because the wakeUp flag will
- * prevent the fetchTask from running.
- * <li>If runningThread is not null, it is after step 2. so we should wakeUp the split reader
- * instead of interrupt the fetcher.
- * </ul>
+ * in the task queue, we should just notify that a task is available. If the fetcher is running
+ * the user split reader, we should call SplitReader.wakeUp() instead.
*
- * <p>The above logic only works in the same {@link #runOnce()} invocation. So we need to
- * synchronize to ensure the wake up logic do not touch a different invocation.
+ * <p>The correctness can be thought of in the following way. The purpose of wake up is to let
+ * the fetcher thread go to the very beginning of the running loop.
*/
void wakeUp(boolean taskOnly) {
// Synchronize to make sure the wake up only works for the current invocation of runOnce().
- synchronized (wakeUp) {
- // Do not wake up repeatedly.
- wakeUp.set(true);
- // Now the wakeUp flag is set.
- SplitFetcherTask currentTask = runningTask;
- if (isRunningTask(currentTask)) {
- // The running task may have missed our wakeUp flag and running, wake it up.
- LOG.debug("Waking up running task {}", currentTask);
- currentTask.wakeUp();
- } else if (!taskOnly) {
- // The task has not started running yet, and it will not run for this
- // runOnce() invocation due to the wakeUp flag. But we might have to
- // wake up the fetcher thread in case it is blocking on the task queue.
- // Only wake up when the thread has started and there is no running task.
- LOG.debug("Waking up fetcher thread.");
- taskQueue.add(WAKEUP_TASK);
- }
+ lock.lock();
+ try {
+ wakeUpUnsafe(taskOnly);
+ } finally {
+ lock.unlock();
}
}
- private void maybeEnqueueTask(SplitFetcherTask task) {
- // Only enqueue unfinished non-fetch task.
- if (!closed.get()
- && isRunningTask(task)
- && task != fetchTask
- && !taskQueue.offerFirst(task)) {
- throw new RuntimeException(
- "The task queue is full. This is only theoretically possible when really bad thing happens.");
- }
- if (task != null) {
- LOG.debug("Enqueued task {}", task);
+ private void wakeUpUnsafe(boolean taskOnly) {
+ assert lock.isHeldByCurrentThread();
Review Comment:
Same here, locking would be an option because of ReentrantLock.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
+ Collection<String> splitsToPause = new ArrayList<>();
+ Collection<String> splitsToResume = new ArrayList<>();
+ splitCurrentWatermarks.forEach(
+ (splitId, splitWatermark) -> {
+ if (splitWatermark > currentMaxDesiredWatermark) {
+ splitsToPause.add(splitId);
+ } else if (currentlyPausedSplits.contains(splitId)) {
+ splitsToResume.add(splitId);
+ }
+ });
+ splitsToPause.removeAll(currentlyPausedSplits);
+ if (!splitsToPause.isEmpty() || !splitsToResume.isEmpty()) {
+ pauseOrResumeSplits(splitsToPause, splitsToResume);
+ currentlyPausedSplits.addAll(splitsToPause);
+ splitsToResume.forEach(currentlyPausedSplits::remove);
+ }
+ }
+
+ private void pauseOrResumeSplits(
+ Collection<String> splitsToPause, Collection<String> splitsToResume) {
+ if (!allowUnalignedSourceSplits || sourceReaderSupportsPauseOrResumeSplits) {
+ try {
+ sourceReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ } catch (UnsupportedOperationException e) {
+ sourceReaderSupportsPauseOrResumeSplits = false;
+ if (!allowUnalignedSourceSplits) {
+ throw new UnsupportedOperationException("", e);
Review Comment:
In my understanding, readers which do not support pausing splits will fail the job, unless the configuration to allow unaligned splits is set. Hence, I don't understand the `sourceReaderSupportsPauseOrResumeSplits` flag.
I suggest the following:
```suggestion
try {
sourceReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
} catch (UnsupportedOperationException e) {
sourceReaderSupportsPauseOrResumeSplits = false;
if (!allowUnalignedSourceSplits) {
throw e;
```
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
Review Comment:
It would be great if we didn't have to special case a single split. I understand that the `SourceReaderBase` will stop polling in this case, but it would be beneficial if there was only one path.
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcher.java:
##########
@@ -111,57 +121,106 @@ public void run() {
splitReader.close();
} catch (Exception e) {
errorHandler.accept(e);
+ } finally {
+ LOG.info("Split fetcher {} exited.", id);
+ // This executes after possible errorHandler.accept(t). If these operations bear
+ // a happens-before relation, then we can checking side effect of
+ // errorHandler.accept(t)
+ // to know whether it happened after observing side effect of shutdownHook.run().
+ shutdownHook.run();
}
- LOG.info("Split fetcher {} exited.", id);
- // This executes after possible errorHandler.accept(t). If these operations bear
- // a happens-before relation, then we can checking side effect of errorHandler.accept(t)
- // to know whether it happened after observing side effect of shutdownHook.run().
- shutdownHook.run();
}
}
/** Package private method to help unit test. */
- void runOnce() {
+ boolean runOnce() {
+ // first blocking call = get next task. blocks only if there are no active splits and queued
+ // tasks.
+ SplitFetcherTask task;
+ lock.lock();
try {
- // The fetch task should run if the split assignment is not empty or there is a split
- // change.
- if (shouldRunFetchTask()) {
- runningTask = fetchTask;
- } else {
- runningTask = taskQueue.take();
+ if (closed) {
+ return false;
}
- // Now the running task is not null. If wakeUp() is called after this point,
- // task.wakeUp() will be called. On the other hand, if the wakeUp() call was make before
- // this point, the wakeUp flag must have already been set. The code hence checks the
- // wakeUp
- // flag first to avoid an unnecessary task run.
- // Note that the runningTask may still encounter the case that the task is waken up
- // before
- // the it starts running.
- LOG.debug("Prepare to run {}", runningTask);
- if (!wakeUp.get() && runningTask.run()) {
- LOG.debug("Finished running task {}", runningTask);
- // the task has finished running. Set it to null so it won't be enqueued.
- runningTask = null;
- checkAndSetIdle();
+
+ task = getNextTaskUnsafe();
+ if (task == null) {
+ // (spurious) wakeup, so just repeat
+ return true;
}
+
+ LOG.debug("Prepare to run {}", task);
+ // store task for #wakeUp
+ this.runningTask = task;
+ } finally {
+ lock.unlock();
+ }
+
+ // execute the task outside of lock, so that it can be woken up
+ boolean taskFinished;
+ try {
+ taskFinished = task.run();
} catch (Exception e) {
throw new RuntimeException(
String.format(
"SplitFetcher thread %d received unexpected exception while polling the records",
id),
e);
}
- // If the task is not null that means this task needs to be re-executed. This only
- // happens when the task is the fetching task or the task was interrupted.
- maybeEnqueueTask(runningTask);
- synchronized (wakeUp) {
- // Set the running task to null. It is necessary for the shutdown method to avoid
- // unnecessarily interrupt the running task.
- runningTask = null;
- // Set the wakeUp flag to false.
- wakeUp.set(false);
- LOG.debug("Cleaned wakeup flag.");
+
+ // re-acquire lock as all post-processing steps, need it
+ lock.lock();
+ try {
+ this.runningTask = null;
+ processTaskResultUnsafe(task, taskFinished);
+ } finally {
+ lock.unlock();
+ }
+ return true;
+ }
+
+ private void processTaskResultUnsafe(SplitFetcherTask task, boolean taskFinished) {
+ assert lock.isHeldByCurrentThread();
Review Comment:
Same as for the other lock asserts.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
+ Collection<String> splitsToPause = new ArrayList<>();
+ Collection<String> splitsToResume = new ArrayList<>();
+ splitCurrentWatermarks.forEach(
+ (splitId, splitWatermark) -> {
+ if (splitWatermark > currentMaxDesiredWatermark) {
+ splitsToPause.add(splitId);
+ } else if (currentlyPausedSplits.contains(splitId)) {
+ splitsToResume.add(splitId);
+ }
+ });
+ splitsToPause.removeAll(currentlyPausedSplits);
+ if (!splitsToPause.isEmpty() || !splitsToResume.isEmpty()) {
+ pauseOrResumeSplits(splitsToPause, splitsToResume);
+ currentlyPausedSplits.addAll(splitsToPause);
+ splitsToResume.forEach(currentlyPausedSplits::remove);
+ }
Review Comment:
What if all splits are paused, do we signal this back to the user like when we only have a single split?
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcher.java:
##########
@@ -210,138 +313,73 @@ Map<String, SplitT> assignedSplits() {
/**
* Package private for unit test.
*
- * @return true if task queue is not empty, false otherwise.
+ * @return true if task queue is empty, false otherwise.
*/
boolean isIdle() {
- return isIdle;
- }
-
- /**
- * Check whether the fetch task should run. The fetch task should only run when all the
- * following conditions are met. 1. there is no task in the task queue to run. 2. there are
- * assigned splits Package private for testing purpose.
- *
- * @return whether the fetch task should be run.
- */
- boolean shouldRunFetchTask() {
- return taskQueue.isEmpty() && !assignedSplits.isEmpty();
+ lock.lock();
+ try {
+ return assignedSplits.isEmpty() && taskQueue.isEmpty() && runningTask == null;
+ } finally {
+ lock.unlock();
+ }
}
/**
* Wake up the fetcher thread. There are only two blocking points in a running fetcher. 1.
- * Taking the next task out of the task queue. 2. Running a task.
+ * Waiting for the next task in an idle fetcher. 2. Running a task.
*
* <p>They need to be waken up differently. If the fetcher is blocking waiting on the next task
- * in the task queue, we should just interrupt the fetcher thread. If the fetcher is running the
- * user split reader, we should call SplitReader.wakeUp() instead of naively interrupt the
- * thread.
- *
- * <p>The correctness can be think of in the following way. The purpose of wake up is to let the
- * fetcher thread go to the very beginning of the running loop. There are three major events in
- * each run of the loop.
- *
- * <ol>
- * <li>pick a task (blocking)
- * <li>assign the task to runningTask variable.
- * <li>run the runningTask. (blocking)
- * </ol>
- *
- * <p>We don't need to worry about things after step 3 because there is no blocking point
- * anymore.
- *
- * <p>We always first set the wakeup flag when waking up the fetcher, then use the value of
- * running task to determine where the fetcher thread is.
- *
- * <ul>
- * <li>If runningThread is null, it is before step 2, so we should interrupt fetcher. This
- * interruption will not be propagated to the split reader, because the wakeUp flag will
- * prevent the fetchTask from running.
- * <li>If runningThread is not null, it is after step 2. so we should wakeUp the split reader
- * instead of interrupt the fetcher.
Review Comment:
Do we want to explain how the new logic works?
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -282,4 +282,21 @@ public enum VertexDescriptionMode {
.withDescription(
"Whether name of vertex includes topological index or not. "
+ "When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default");
+
+ @PublicEvolving
+ public static final ConfigOption<Boolean> ALLOW_UNALIGNED_SOURCE_SPLITS =
+ key("pipeline.watermark-alignment.allow-unaligned-source-splits")
+ .booleanType()
+ .defaultValue(false)
Review Comment:
Just a thought: Do we have to have another configuration option? I think we could just keep the default behavior prior adding this feature and not do split alignment. We could print a WARN when this happens.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
Review Comment:
We should consider whether we need the eager pausing logic here. There is the periodic `checkSplitWatermarkAlignment` method which achieves the same. The default with one split was to pause at the alignment interval only, so maybe we should keep that behavior.
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/PauseOrResumeSplitsTask.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.base.source.reader.fetcher;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collection;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Changes the paused splits of a n{@link SplitReader}. The task is used by default in {@link
Review Comment:
"a n" typo?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -175,6 +188,9 @@ private enum OperatingMode {
private @Nullable LatencyMarkerEmitter<OUT> latencyMarkerEmitter;
+ private final boolean allowUnalignedSourceSplits;
+ private boolean sourceReaderSupportsPauseOrResumeSplits = true;
Review Comment:
IMHO this flag is not needed.
```suggestion
```
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcher.java:
##########
@@ -171,17 +230,54 @@ void runOnce() {
* @param splitsToAdd the splits to add.
*/
public void addSplits(List<SplitT> splitsToAdd) {
- enqueueTask(new AddSplitsTask<>(splitReader, splitsToAdd, assignedSplits));
- wakeUp(true);
+ lock.lock();
+ try {
+ enqueueTaskUnsafe(new AddSplitsTask<>(splitReader, splitsToAdd, assignedSplits));
+ wakeUpUnsafe(true);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Called when some splits of this source instance progressed too much beyond the global
+ * watermark of all subtasks. If the split reader implements {@link SplitReader}, it will relay
+ * the information asynchronously through the split fetcher thread.
Review Comment:
This JavaDoc leaks information about potential callers but does not explain what this method actually does.
##########
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/fetcher/KafkaSourceFetcherManager.java:
##########
@@ -68,7 +69,7 @@ public KafkaSourceFetcherManager(
Supplier<SplitReader<ConsumerRecord<byte[], byte[]>, KafkaPartitionSplit>>
splitReaderSupplier,
Consumer<Collection<String>> splitFinishedHook) {
- super(elementsQueue, splitReaderSupplier, splitFinishedHook);
+ super(elementsQueue, splitReaderSupplier, new Configuration(), splitFinishedHook);
Review Comment:
Why do we need pass in the correct configuration?
##########
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java:
##########
@@ -179,6 +180,21 @@ public void handleSplitsChanges(SplitsChange<PulsarPartitionSplit> splitsChanges
this.pulsarConsumer = consumer;
}
+ @Override
+ public void pauseOrResumeSplits(
+ Collection<PulsarPartitionSplit> splitsToPause,
+ Collection<PulsarPartitionSplit> splitsToResume) {
+ if (splitsToPause.size() > 1 || splitsToResume.size() > 1) {
+ throw new IllegalStateException("This pulsar split reader only support one split.");
Review Comment:
```suggestion
throw new IllegalStateException("This pulsar split reader only supports one split.");
```
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
Review Comment:
Is this operation safe? We are adding new splits here while at the same time this set is cleared in ` checkSplitWatermarkAlignment` (see https://github.com/apache/flink/pull/20485/files#diff-eba14821fb3e96f6f20e3116ceab893c2f18cff605b177dad485aadc43ac4f56R619 line 619).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r960705416
##########
docs/layouts/shortcodes/generated/pipeline_configuration.html:
##########
@@ -128,5 +128,11 @@
<td>Boolean</td>
<td>Whether name of vertex includes topological index or not. When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default</td>
</tr>
+ <tr>
+ <td><h5>pipeline.watermark-alignment.allow-unaligned-source-splits</h5></td>
Review Comment:
IMHO something like `source.reader.synchronize.splits` would be more understandable for users.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
dawidwys commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1246745732
Thank you for taking over! @mxm
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1245573375
Tests are passing now.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r963034437
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -282,4 +282,21 @@ public enum VertexDescriptionMode {
.withDescription(
"Whether name of vertex includes topological index or not. "
+ "When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default");
+
+ @PublicEvolving
+ public static final ConfigOption<Boolean> ALLOW_UNALIGNED_SOURCE_SPLITS =
+ key("pipeline.watermark-alignment.allow-unaligned-source-splits")
+ .booleanType()
+ .defaultValue(false)
Review Comment:
I think we are on the same page. With only FLIP 217, a slow and a fast split in the same operator instance will pause both the fast and the slow split. That likely means the gap between the two splits will not become smaller but bigger or remain the same. The lag behavior is very much random.
At the least, we should provide good error reporting in case a user wants to use a source without split alignment. The error report should also inform the user how to override this behavior.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r963759026
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/splitreader/SplitReader.java:
##########
@@ -68,4 +69,29 @@
* @throws Exception if closing the split reader failed.
*/
void close() throws Exception;
+
+ /**
+ * Pauses or resumes reading of individual splits readers.
+ *
+ * <p>Note that no other methods can be called in parallel, so it's fine to non-atomically
+ * update subscriptions. This method is simply providing connectors with more expressive APIs
+ * the opportunity to update all subscriptions at once.
+ *
+ * <p>This is currently used to align the watermarks of splits, if watermark alignment is used
+ * and the source reads from more than one split.
+ *
+ * <p>The default implementation throws an {@link UnsupportedOperationException} where the
+ * default implementation will be removed in future releases. To be compatible with future
+ * releases, it is recommended to implement this method and override the default implementation.
+ *
+ * @param splitsToPause the splits to pause
+ * @param splitsToResume the splits to resume
+ */
+ default void pauseOrResumeSplits(
+ Collection<SplitT> splitsToPause, Collection<SplitT> splitsToResume) {
+ throw new UnsupportedOperationException(
+ "This split reader does not support pause or resume. (To use watermark alignment "
+ + "and allow unaligned source splits set configuration parameter "
+ + "'pipeline.watermark-alignment.allow-unaligned-source-splits' to true.)");
Review Comment:
Thanks, have modified it slightly:
```java
throw new UnsupportedOperationException(
"This split reader does not support pausing or resuming splits which can lead to unaligned splits.\n"
+ "Unaligned splits are splits where the output watermarks of the splits have diverged more than the allowed limit.\n"
+ "It is highly discouraged to use unaligned source splits, as this leads to unpredictable\n"
+ "watermark alignment if there is more than a single split. It is recommended to implement pausing splits\n"
+ "for this source. At your own risk, you can allow unaligned source splits by setting the\n"
+ "configuration parameter `pipeline.watermark-alignment.allow-unaligned-source-splits' to true.\n"
+ "Beware that this configuration parameter will be dropped in a future Flink release.");
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r963343155
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReader.java:
##########
@@ -132,4 +139,28 @@ default void handleSourceEvents(SourceEvent sourceEvent) {}
*/
@Override
default void notifyCheckpointComplete(long checkpointId) throws Exception {}
+
+ /**
+ * Pauses or resumes reading of individual source splits.
+ *
+ * <p>Note that no other methods can be called in parallel, so updating subscriptions can be
+ * done atomically. This method is simply providing connectors with more expressive APIs the
+ * opportunity to update all subscriptions at once.
+ *
+ * <p>This is currently used to align the watermarks of splits, if watermark alignment is used
+ * and the source reads from more than one split.
+ *
+ * <p>The default implementation throws an {@link UnsupportedOperationException} where the
+ * default implementation will be removed in future releases. To be compatible with future
+ * releases, it is recommended to implement this method and override the default implementation.
+ *
+ * @param splitsToPause the splits to pause
+ * @param splitsToResume the splits to resume
+ */
+ @PublicEvolving
+ default void pauseOrResumeSplits(
+ Collection<String> splitsToPause, Collection<String> splitsToResume) {
+ throw new UnsupportedOperationException(
+ "This source reader does not support pause or resume splits.");
Review Comment:
Shouldn't this be kept in sync with the other similar `throw new UnsupportedOperationException` in `SplitReader`?
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/splitreader/SplitReader.java:
##########
@@ -68,4 +69,29 @@
* @throws Exception if closing the split reader failed.
*/
void close() throws Exception;
+
+ /**
+ * Pauses or resumes reading of individual splits readers.
+ *
+ * <p>Note that no other methods can be called in parallel, so it's fine to non-atomically
+ * update subscriptions. This method is simply providing connectors with more expressive APIs
+ * the opportunity to update all subscriptions at once.
+ *
+ * <p>This is currently used to align the watermarks of splits, if watermark alignment is used
+ * and the source reads from more than one split.
+ *
+ * <p>The default implementation throws an {@link UnsupportedOperationException} where the
+ * default implementation will be removed in future releases. To be compatible with future
+ * releases, it is recommended to implement this method and override the default implementation.
+ *
+ * @param splitsToPause the splits to pause
+ * @param splitsToResume the splits to resume
+ */
+ default void pauseOrResumeSplits(
+ Collection<SplitT> splitsToPause, Collection<SplitT> splitsToResume) {
+ throw new UnsupportedOperationException(
+ "This split reader does not support pause or resume. (To use watermark alignment "
+ + "and allow unaligned source splits set configuration parameter "
+ + "'pipeline.watermark-alignment.allow-unaligned-source-splits' to true.)");
Review Comment:
> This split reader does not support pause or resume. It is highly discourage to use
> watermark alignment and allow unaligned source splits, as this leads to unpredictable
> effectiveness of the watermark alignment. It is recommended to implement pausing splits
> for this source. Temporarily you can allow unaligned source splits by setting
> configuration parameter `pipeline.watermark-alignment.allow-unaligned-source-splits' to true.
> Beware that this configuration parameter will be dropped in a future Flink release.
?
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -282,4 +282,21 @@ public enum VertexDescriptionMode {
.withDescription(
"Whether name of vertex includes topological index or not. "
+ "When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default");
+
+ @PublicEvolving
+ public static final ConfigOption<Boolean> ALLOW_UNALIGNED_SOURCE_SPLITS =
+ key("pipeline.watermark-alignment.allow-unaligned-source-splits")
+ .booleanType()
+ .defaultValue(false)
Review Comment:
I fully agree. I've left a suggestion in this regard.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r952493768
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcher.java:
##########
@@ -171,17 +230,54 @@ void runOnce() {
* @param splitsToAdd the splits to add.
*/
public void addSplits(List<SplitT> splitsToAdd) {
- enqueueTask(new AddSplitsTask<>(splitReader, splitsToAdd, assignedSplits));
- wakeUp(true);
+ lock.lock();
+ try {
+ enqueueTaskUnsafe(new AddSplitsTask<>(splitReader, splitsToAdd, assignedSplits));
+ wakeUpUnsafe(true);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Called when some splits of this source instance progressed too much beyond the global
+ * watermark of all subtasks. If the split reader implements {@link SplitReader}, it will relay
+ * the information asynchronously through the split fetcher thread.
Review Comment:
This JavaDoc leaks information about potential callers but does not explain what this method actually does.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r965717885
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -589,10 +589,6 @@ public void updateCurrentEffectiveWatermark(long watermark) {
@Override
public void updateCurrentSplitWatermark(String splitId, long watermark) {
splitCurrentWatermarks.put(splitId, watermark);
- if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
- pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
- currentlyPausedSplits.add(splitId);
- }
Review Comment:
Yes, I didn't remove this but I fixed a bug which @mas-chen pointed out which would pause the split when there is just a single split.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1241135990
> @mxm Thanks a lot also from my side for you contributions. However, @pnowojski this PR was not orphaned! (I wrote to you that I'm on vacations.)
Yes, sorry for the mental shortcut. There was a delay in the communication between first my vacations and then yours. Around the time @mxm was asking how to continue with this PR, you @smattheis were not responding, and I got a message from you that you can work on it only after @mxm has already volunteered to take it over. After that I didn't want to create more confusion. But sorry for my inaccurate statement that it was "orphaned" :)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r965659207
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReader.java:
##########
@@ -161,6 +161,12 @@ default void notifyCheckpointComplete(long checkpointId) throws Exception {}
default void pauseOrResumeSplits(
Collection<String> splitsToPause, Collection<String> splitsToResume) {
throw new UnsupportedOperationException(
- "This source reader does not support pause or resume splits.");
+ "This split reader does not support pausing or resuming splits which can lead to unaligned splits.\n"
Review Comment:
nit: `This source reader`?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -589,10 +589,6 @@ public void updateCurrentEffectiveWatermark(long watermark) {
@Override
public void updateCurrentSplitWatermark(String splitId, long watermark) {
splitCurrentWatermarks.put(splitId, watermark);
- if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
- pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
- currentlyPausedSplits.add(splitId);
- }
Review Comment:
Why are we removing this? If suddenly split wakes up and starts sending huge amount of records, lack of this code can cause problems.
edit: oh, you have reverted this change later, haven't you?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1241039980
Thank you @pnowojski for your thoughtful comments and guidance. And of course thanks to the original authors @AHeise @dawidwys @smattheis.
Whoops @smattheis I had assumed you wouldn't continue on this PR but I think no harm has been done. Let me know if anything is open on your side. It still need to revise the code with respect to a test failure.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1240740217
> I wouldn't squash all commits into one. The original commits authored by Sebastian, Dawid and Arvid I think are nicely separated. But sure, squashing your fixups to those commits and co-authoring them is also perfectly fine.
If the commits are nicely separated, why were they not handled in separate PRs? Anyhow, I'm ok with just putting my changes in a single commit on top of those.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1243534735
Thanks @pnowojski. Will fix those.
On another note, do you think it makes sense to add a configuration option to disable using split alignment entirely? I'm a bit worried this feature might cause issues with the already integrated Kafka/Pulsar sources that we are not foreseeing right now. The added configuration option only allows to disable split alignments for non-compatible connectors. We could add another option to disable it entirely to retain the old behavior (disabled by default).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r960476613
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
Review Comment:
We do modify `currentlyPausedSplits` two lines down: https://github.com/apache/flink/pull/20485/files#diff-eba14821fb3e96f6f20e3116ceab893c2f18cff605b177dad485aadc43ac4f56R622
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm merged pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm merged PR #20485:
URL: https://github.com/apache/flink/pull/20485
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1249094709
Yes, it took longer than it should have. Good that it's finally been merged ๐
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r951456255
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/PauseOrResumeSplitsTask.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.base.source.reader.fetcher;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collection;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Changes the paused splits of a n{@link SplitReader}. The task is used by default in {@link
+ * SplitFetcherManager} and assumes that a {@link SplitFetcher} has multiple splits. For {@code
+ * SplitFetchers} with single splits, it's instead recommended to subclass {@link
+ * SplitFetcherManager} and pause the whole {@code SplitFetcher}.
+ *
+ * @param <SplitT> the type of the split
+ */
+@Internal
+class PauseOrResumeSplitsTask<SplitT extends SourceSplit> implements SplitFetcherTask {
+ private static final Logger LOG = LoggerFactory.getLogger(PauseOrResumeSplitsTask.class);
+ private final SplitReader<?, SplitT> splitReader;
+ private final Collection<SplitT> splitsToPause;
+ private final Collection<SplitT> splitsToResume;
+ private final boolean allowUnalignedSourceSplits;
+
+ PauseOrResumeSplitsTask(
+ SplitReader<?, SplitT> splitReader,
+ Collection<SplitT> splitsToPause,
+ Collection<SplitT> splitsToResume,
+ boolean allowUnalignedSourceSplits) {
+ this.splitReader = checkNotNull(splitReader);
+ this.splitsToPause = checkNotNull(splitsToPause);
+ this.splitsToResume = checkNotNull(splitsToResume);
+ this.allowUnalignedSourceSplits = allowUnalignedSourceSplits;
+ }
+
+ @Override
+ public boolean run() throws IOException {
+ try {
+ splitReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ } catch (UnsupportedOperationException e) {
+ if (!allowUnalignedSourceSplits) {
+ throw new UnsupportedOperationException("", e);
Review Comment:
+1 for no wrapping.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
+ Collection<String> splitsToPause = new ArrayList<>();
+ Collection<String> splitsToResume = new ArrayList<>();
+ splitCurrentWatermarks.forEach(
+ (splitId, splitWatermark) -> {
+ if (splitWatermark > currentMaxDesiredWatermark) {
+ splitsToPause.add(splitId);
+ } else if (currentlyPausedSplits.contains(splitId)) {
+ splitsToResume.add(splitId);
+ }
+ });
+ splitsToPause.removeAll(currentlyPausedSplits);
+ if (!splitsToPause.isEmpty() || !splitsToResume.isEmpty()) {
+ pauseOrResumeSplits(splitsToPause, splitsToResume);
+ currentlyPausedSplits.addAll(splitsToPause);
+ splitsToResume.forEach(currentlyPausedSplits::remove);
+ }
+ }
+
+ private void pauseOrResumeSplits(
+ Collection<String> splitsToPause, Collection<String> splitsToResume) {
+ if (!allowUnalignedSourceSplits || sourceReaderSupportsPauseOrResumeSplits) {
+ try {
+ sourceReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ } catch (UnsupportedOperationException e) {
+ sourceReaderSupportsPauseOrResumeSplits = false;
+ if (!allowUnalignedSourceSplits) {
+ throw new UnsupportedOperationException("", e);
Review Comment:
+1 for no wrapping.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1231242878
> I wonder about the way the SourceOperator calls through the stack via pauseOrResumeSplits while also being called back with watermarks. My intuition would have been to do the pausing directly at the splits based on the current max watermark lag, e.g. by putting splits into a suspended mode where no more output is accepted from a split. I guess this could be tricky because the implementation of the source doesn't necessarily know it's paused and could timeout or otherwise misbehave. So it looks like the current implementation is simpler.
Currently Splits are just pojos, data carriers, without much logic. If I understand you correctly, what you are suggesting would require us to push some availability logic down to the splits and it would still require handling of this availability in the `SourceReader` - basically the same logic that we have in the current proposal. The difference would be only that `SourceReader` would be informed about (un)availability change of a split, from the split, instead of from the `pauseOrResume` call.
Given that most of the users are expected to just use `SourceReaderBase`, the current proposal gives them watermark alignment for free. While adding availability logic to the splits would increase exposure of this feature in the public API, without much benefit?
> Any thoughts on how and when to proceed with this PR? We'd like to help to bring this over the finish line.
Indeed there is an issue now, that I think @smattheis doesn't work on Flink anymore, so he can not finish this PR. I will go through the PR myself and do a quick pass/respond to the inlined questions from you @mxm and @mas-chen , but it would be great if someone could take over applying the actual changes after the review. Would you be up to do that @mxm ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r964443840
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/splitreader/SplitReader.java:
##########
@@ -68,4 +69,29 @@
* @throws Exception if closing the split reader failed.
*/
void close() throws Exception;
+
+ /**
+ * Pauses or resumes reading of individual splits readers.
+ *
+ * <p>Note that no other methods can be called in parallel, so it's fine to non-atomically
+ * update subscriptions. This method is simply providing connectors with more expressive APIs
+ * the opportunity to update all subscriptions at once.
+ *
+ * <p>This is currently used to align the watermarks of splits, if watermark alignment is used
+ * and the source reads from more than one split.
+ *
+ * <p>The default implementation throws an {@link UnsupportedOperationException} where the
+ * default implementation will be removed in future releases. To be compatible with future
+ * releases, it is recommended to implement this method and override the default implementation.
+ *
+ * @param splitsToPause the splits to pause
+ * @param splitsToResume the splits to resume
+ */
+ default void pauseOrResumeSplits(
+ Collection<SplitT> splitsToPause, Collection<SplitT> splitsToResume) {
+ throw new UnsupportedOperationException(
+ "This split reader does not support pause or resume. (To use watermark alignment "
+ + "and allow unaligned source splits set configuration parameter "
+ + "'pipeline.watermark-alignment.allow-unaligned-source-splits' to true.)");
Review Comment:
Sounds great!
> "watermark alignment if there is more than a single split."
change to:
> "watermark alignment if there is more than a single split per split reader."
?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r964825458
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
+ }
+
+ /**
+ * Finds the splits that are beyond the current max watermark and pauses them. At the same time,
+ * splits that have been paused and where the global watermark caught up are resumed.
+ *
+ * <p>Note: This takes effect only if there are multiple splits, otherwise it does nothing.
+ */
+ private void checkSplitWatermarkAlignment() {
+ if (numSplits <= 1) {
+ return; // If there is only a single split, we do not pause the split but the source.
+ }
Review Comment:
I'll keep this as-is for now.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] smattheis commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
smattheis commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1241738820
@mxm @pnowojski All good. Thanks for driving it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r965066688
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherPauseResumeSplitReaderTest.java:
##########
@@ -0,0 +1,311 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.base.source.reader.fetcher;
+
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.mocks.MockSourceReader;
+import org.apache.flink.connector.base.source.reader.mocks.MockSplitReader;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.core.testutils.CommonTestUtils;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests {@link SplitFetcher} integration to pause or resume {@link SplitReader} based on {@link
+ * SourceReader} output.
+ */
+public class SplitFetcherPauseResumeSplitReaderTest {
+
+ /**
+ * Tests if pause or resume shows expected behavior which requires creation and execution of
+ * {@link SplitFetcher} tasks.
+ *
+ * @throws Exception on error.
+ */
+ @ParameterizedTest(name = "Individual reader per split: {0}")
+ @ValueSource(booleans = {false, true})
+ public void testPauseResumeSplitReaders(boolean individualReader) throws Exception {
+ final AtomicInteger numSplitReaders = new AtomicInteger();
+ final MockSplitReader.Builder readerBuilder =
+ SteppingSourceReaderTestHarness.createSplitReaderBuilder();
+ final SteppingSourceReaderTestHarness testHarness =
+ new SteppingSourceReaderTestHarness(
+ () -> {
+ numSplitReaders.getAndIncrement();
+ return readerBuilder.build();
+ },
+ new Configuration());
+
+ if (individualReader) {
+ testHarness.addPrefilledSplitsIndividualReader(2, 5);
+ assertThat(numSplitReaders.get()).isEqualTo(2);
+ } else {
+ testHarness.addPrefilledSplitsSingleReader(2, 5);
+ assertThat(numSplitReaders.get()).isEqualTo(1);
+ }
+
+ TestingReaderOutput output = new TestingReaderOutput<>();
+ testHarness.runUntilRecordsEmitted(output, 10, 2);
+ Set<Integer> recordSet = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet).containsExactlyInAnyOrder(new Integer[] {0, 1});
+
+ testHarness.pauseOrResumeSplits(Collections.singleton("0"), Collections.emptyList());
+
+ testHarness.runUntilRecordsEmitted(output, 10, 5);
+ Set<Integer> recordSet2 = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet2).containsExactlyInAnyOrder(new Integer[] {0, 1, 3, 5, 7});
+
+ testHarness.pauseOrResumeSplits(Collections.emptyList(), Collections.singleton("0"));
+
+ testHarness.runUntilAllRecordsEmitted(output, 10);
+ Set<Integer> recordSet3 = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet3)
+ .containsExactlyInAnyOrder(new Integer[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
+ }
+
+ /**
+ * Tests if pause or resume shows expected behavior in case of {@link SplitReader} that does not
+ * support split or resume for scenarios (1) allowed and (2) not allowed.
+ *
+ * @throws Exception on error.
+ */
+ @ParameterizedTest(name = "Allow unaligned source splits: {0}")
+ @ValueSource(booleans = {true, false})
+ public void testPauseResumeUnsupported(boolean allowUnalignedSourceSplits) throws Exception {
+ final AtomicInteger numSplitReaders = new AtomicInteger();
+ final Configuration configuration = new Configuration();
+ configuration.setBoolean(
+ "pipeline.watermark-alignment.allow-unaligned-source-splits",
+ allowUnalignedSourceSplits);
+ final MockSplitReader.Builder readerBuilder =
+ SteppingSourceReaderTestHarness.createSplitReaderBuilder();
+
+ final SteppingSourceReaderTestHarness testHarness =
+ new SteppingSourceReaderTestHarness(
+ () -> {
+ if (numSplitReaders.getAndIncrement() == 0) {
+ return MockSplitReaderUnsupportedPause.cloneBuilder(readerBuilder)
+ .build();
+ } else {
+ return readerBuilder.build();
+ }
+ },
+ configuration);
+
+ testHarness.addPrefilledSplitsIndividualReader(2, 5);
+ assertThat(numSplitReaders.get()).isEqualTo(2);
+
+ TestingReaderOutput output = new TestingReaderOutput<>();
+ testHarness.runUntilRecordsEmitted(output, 10, 2);
+ Set<Integer> recordSet = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet).containsExactlyInAnyOrder(new Integer[] {0, 1});
+
+ testHarness.pauseOrResumeSplits(Collections.singleton("1"), Collections.emptyList());
+
+ testHarness.runUntilRecordsEmitted(output, 10, 5);
+ Set<Integer> recordSet2 = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet2).containsExactlyInAnyOrder(new Integer[] {0, 1, 2, 4, 6});
+
+ testHarness.pauseOrResumeSplits(Collections.singleton("0"), Collections.singleton("1"));
+
+ if (allowUnalignedSourceSplits) {
+ testHarness.runUntilAllRecordsEmitted(output, 10);
+ Set<Integer> recordSet3 = new HashSet<>(output.getEmittedRecords());
+ assertThat(recordSet3)
+ .containsExactlyInAnyOrder(new Integer[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
+ } else {
+ assertThatThrownBy(() -> testHarness.runUntilAllRecordsEmitted(output, 10))
+ .isInstanceOf(RuntimeException.class)
+ .hasCauseInstanceOf(UnsupportedOperationException.class);
+ }
+ }
+
+ private static class MockSteppingSplitFetcherManager<E, SplitT extends SourceSplit>
+ extends SingleThreadFetcherManager<E, SplitT> {
+
+ public MockSteppingSplitFetcherManager(
+ FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
+ Supplier<SplitReader<E, SplitT>> splitReaderSupplier,
+ Configuration configuration) {
+ super(elementsQueue, splitReaderSupplier, configuration);
+ }
+
+ @Override
+ public void addSplits(List<SplitT> splitsToAdd) {
+ SplitFetcher<E, SplitT> fetcher = createSplitFetcher();
+ fetcher.addSplits(splitsToAdd);
+ }
+
+ public void runEachOnce() {
+ for (SplitFetcher<E, SplitT> fetcher : fetchers.values()) {
+ fetcher.runOnce();
+ }
+ }
+ }
+
+ private static class MockSplitReaderUnsupportedPause extends MockSplitReader {
+ public MockSplitReaderUnsupportedPause(
+ int numRecordsPerSplitPerFetch,
+ boolean separatedFinishedRecord,
+ boolean blockingFetch) {
+ super(numRecordsPerSplitPerFetch, separatedFinishedRecord, blockingFetch);
+ }
+
+ @Override
+ public void pauseOrResumeSplits(
+ Collection<MockSourceSplit> splitsToPause,
+ Collection<MockSourceSplit> splitsToResume) {
+ throw new UnsupportedOperationException();
+ }
+
+ public static class Builder extends MockSplitReader.Builder {
+ public Builder(MockSplitReader.Builder other) {
+ super(other);
+ }
+
+ @Override
+ public MockSplitReader build() {
+ return new MockSplitReaderUnsupportedPause(
+ numRecordsPerSplitPerFetch, separatedFinishedRecord, blockingFetch);
+ }
+ }
+
+ public static Builder cloneBuilder(MockSplitReader.Builder other) {
+ return new MockSplitReaderUnsupportedPause.Builder(other);
+ }
+ }
+
+ private static class SteppingSourceReaderTestHarness {
+ private final MockSteppingSplitFetcherManager<int[], MockSourceSplit> fetcherManager;
+ private final MockSourceReader sourceReader;
+
+ public SteppingSourceReaderTestHarness(
+ Supplier<SplitReader<int[], MockSourceSplit>> splitReaderSupplier,
+ Configuration configuration) {
+ FutureCompletingBlockingQueue<RecordsWithSplitIds<int[]>> queue =
+ new FutureCompletingBlockingQueue<>(10);
+ this.fetcherManager =
+ new MockSteppingSplitFetcherManager<>(
+ queue, splitReaderSupplier, configuration);
+ this.sourceReader =
+ new MockSourceReader(
+ queue, fetcherManager, configuration, new TestingReaderContext());
+ }
+
+ private static List<MockSourceSplit> createPrefilledSplits(int numSplits, int numRecords) {
+ final List<MockSourceSplit> splits = new ArrayList<>(numSplits);
+ for (int splitId = 0; splitId < numSplits; splitId++) {
+ MockSourceSplit split = new MockSourceSplit(splitId, 0, numRecords);
+ for (int i = 0; i < numRecords; i++) {
+ split.addRecord(i * numSplits + splitId);
+ }
+ splits.add(split);
+ }
+ return splits;
+ }
+
+ public void addPrefilledSplitsSingleReader(int numSplits, int numRecords) {
+ sourceReader.addSplits(createPrefilledSplits(numSplits, numRecords));
+ sourceReader.notifyNoMoreSplits();
Review Comment:
This is not required. The signal simply allows for a clean shutdown of the source.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r965983501
##########
flink-core/src/main/java/org/apache/flink/api/common/eventtime/CombinedWatermarkStatus.java:
##########
@@ -95,10 +97,12 @@ public boolean updateCombinedWatermark() {
static class PartialWatermark {
private long watermark = Long.MIN_VALUE;
private boolean idle = false;
+
+ @Nullable
Review Comment:
Agreed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1240605079
> I'd prefer this to be a single suqashed commit with the use of Co-Authored-By. IMHO there is no benefit in merging the individual commits because they are very much tight to each other. This keeps the history clean while retaining authorship. What do you think?
I wouldn't squash all commits into one. The original commits authored by Sebastian, Dawid and Arvid I think are nicely separated. But sure, squashing your fixups to those commits and co-authoring them is also perfectly fine.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1245274821
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1241967274
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1245270504
Fair point, users may modify the job to disable alignment. It doesn't make sense to override the watermark alignment settings from the config.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on PR #20485:
URL: https://github.com/apache/flink/pull/20485#issuecomment-1230251521
Any thoughts on how and when to proceed with this PR? We'd like to help to bring this over the finish line.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r960504503
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -282,4 +282,21 @@ public enum VertexDescriptionMode {
.withDescription(
"Whether name of vertex includes topological index or not. "
+ "When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default");
+
+ @PublicEvolving
+ public static final ConfigOption<Boolean> ALLOW_UNALIGNED_SOURCE_SPLITS =
+ key("pipeline.watermark-alignment.allow-unaligned-source-splits")
+ .booleanType()
+ .defaultValue(false)
Review Comment:
I've read the mailing list thread. I still think it would be better to not enforce this feature for now. If I understand correctly, watermark alignment without split alignment still works but it is possible for splits to overtake each other which can slow down the processing due to regular watermark alignment. I would take it to the mailing list if we want to revisit this decision.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r952590047
##########
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/fetcher/KafkaSourceFetcherManager.java:
##########
@@ -68,7 +69,7 @@ public KafkaSourceFetcherManager(
Supplier<SplitReader<ConsumerRecord<byte[], byte[]>, KafkaPartitionSplit>>
splitReaderSupplier,
Consumer<Collection<String>> splitFinishedHook) {
- super(elementsQueue, splitReaderSupplier, splitFinishedHook);
+ super(elementsQueue, splitReaderSupplier, new Configuration(), splitFinishedHook);
Review Comment:
Should we pass in the correct configuration?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r963060391
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
Review Comment:
Thanks! I was worried about threading issues but I see that it's safe due to the single-threaded execution model.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r965853207
##########
flink-core/src/main/java/org/apache/flink/api/common/eventtime/CombinedWatermarkStatus.java:
##########
@@ -95,10 +97,12 @@ public boolean updateCombinedWatermark() {
static class PartialWatermark {
private long watermark = Long.MIN_VALUE;
private boolean idle = false;
+
+ @Nullable
Review Comment:
If this is only used for the sake of tests, I would actually prefer to leave this field as non-null with no-op lambda functions.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] mxm commented on a diff in pull request #20485: [FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits
Posted by GitBox <gi...@apache.org>.
mxm commented on code in PR #20485:
URL: https://github.com/apache/flink/pull/20485#discussion_r964964213
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -559,14 +577,67 @@ private void createOutputForSplits(List<SplitT> newSplits) {
private void updateMaxDesiredWatermark(WatermarkAlignmentEvent event) {
currentMaxDesiredWatermark = event.getMaxWatermark();
+ checkSplitWatermarkAlignment();
sourceMetricGroup.updateMaxDesiredWatermark(currentMaxDesiredWatermark);
}
- private void onWatermarkEmitted(long emittedWatermark) {
- lastEmittedWatermark = emittedWatermark;
+ @Override
+ public void updateCurrentEffectiveWatermark(long watermark) {
+ lastEmittedWatermark = watermark;
checkWatermarkAlignment();
}
+ @Override
+ public void updateCurrentSplitWatermark(String splitId, long watermark) {
+ splitCurrentWatermarks.put(splitId, watermark);
+ if (currentMaxDesiredWatermark < watermark && !currentlyPausedSplits.contains(splitId)) {
+ pauseOrResumeSplits(Collections.singletonList(splitId), Collections.emptyList());
+ currentlyPausedSplits.add(splitId);
+ }
Review Comment:
Keeping the logic because it's helpful to prevent too much divergence in case the watermark alignment interval is high.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org