This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
The following commit(s) were added to refs/heads/main by this push:
new 07d9375a [FLINK-40779][connector-jdbc] Snapshot splitter state
together with its splits
07d9375a is described below
commit 07d9375a6aa2df86c96bb1cef6186f76a4b33dc4
Author: yongfu.gao <[email protected]>
AuthorDate: Mon Sep 28 22:44:29 2026 +0800
[FLINK-40779][connector-jdbc] Snapshot splitter state together with its
splits
JdbcSourceEnumerator enumerates splits through callAsync, so the splitter
cursor advances on a worker thread while the produced splits only reach
`unassigned` in the handler on the coordinator thread. Reading the splitter
state directly in snapshotState() could therefore record a state that had
already moved past splits that were still in flight; those splits were never
re-enumerated after a restore.
Capture the splits and the splitter state together inside the callable, and
commit the state from the handler. Enumerations are limited to one in
flight,
so the committed state and the buffered splits stay in sync no matter how
callAsync schedules the callables; relying on their completion order would
not
be safe. snapshotState() now uses that committed state.
For the setSql path the state is also seeded from the provider when no
state is
configured. Stateful providers such as JdbcSlideTimingParameterProvider are
mutated in place while enumerating, so a first-batch checkpoint taken with a
null state could not roll the provider back on restore and the in-flight
first
batch was still lost.
The fix sits in the enumerator rather than in the individual
SplitterEnumerator implementations so that every implementation is covered
(SlideTimingSplitterEnumerator is exposed the same way) and the
PublicEvolving interface stays unchanged.
On the error path the committed state lags behind, so a restore may
re-enumerate splits that were already produced. That is at-least-once rather
than data loss.
Generated-by: DeepSeek Harness (deepseek-v4.1-flash)
---
.../source/enumerator/JdbcSourceEnumerator.java | 69 ++++-
.../enumerator/SqlTemplateSplitEnumerator.java | 14 +-
.../JdbcSourceEnumeratorCheckpointRaceTest.java | 325 +++++++++++++++++++++
3 files changed, 399 insertions(+), 9 deletions(-)
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumerator.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumerator.java
index d371fd9f..f7f120f5 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumerator.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumerator.java
@@ -32,6 +32,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
@@ -51,6 +52,19 @@ public class JdbcSourceEnumerator
private final List<Integer> readersWaitingForSplits = new ArrayList<>();
private final AtomicInteger asyncCallsPending = new AtomicInteger(0);
+ /**
+ * The splitter state that corresponds to the splits that have already
been handed over to
+ * {@link #unassigned} or to a reader.
+ *
+ * <p>Only touched from the coordinator thread: {@link
#onSplitsDiscovered} commits it and
+ * {@link #snapshotState(long)} reads it. Both run on the same thread as
each other, so the two
+ * never interleave. It must NOT be read from {@link
SplitterEnumerator#serializableState()}
+ * directly in {@link #snapshotState(long)}: enumeration happens on a
worker thread, so that
+ * state can already have advanced past splits that are still in flight
and therefore not part
+ * of the checkpoint.
+ */
+ private Serializable lastHandledSplitterState;
+
public JdbcSourceEnumerator(
SplitEnumeratorContext<JdbcSourceSplit> context,
SplitterEnumerator splitterEnumerator,
@@ -65,6 +79,9 @@ public class JdbcSourceEnumerator
@Override
public void start() {
splitterEnumerator.start(connectionProvider);
+ // No split has been handled yet: this is the state a concurrent
checkpoint has to fall
+ // back to while the first batch is still being enumerated on a worker
thread.
+ lastHandledSplitterState = splitterEnumerator.serializableState();
preDiscoverSplits();
}
@@ -120,7 +137,7 @@ public class JdbcSourceEnumerator
Collections.emptyList(),
Collections.emptyList(),
new ArrayList<>(unassigned),
- splitterEnumerator.serializableState());
+ lastHandledSplitterState);
}
private Optional<JdbcSourceSplit> getNextSplit() {
@@ -134,17 +151,31 @@ public class JdbcSourceEnumerator
}
private void preDiscoverSplits() {
- int targetParallelism = context.currentParallelism();
- while (asyncCallsPending.get() < targetParallelism
- && !splitterEnumerator.isAllSplitsFinished()) {
+ // Keep at most one enumeration in flight. The splitter state that
goes with a batch is
+ // committed by its handler, so the commits only stay ordered
consistently with the state
+ // advances while there is a single enumeration running at a time.
callAsync allows
+ // concurrent callables, so relying on the completion order would be
unsafe.
+ while (asyncCallsPending.get() < 1 &&
!splitterEnumerator.isAllSplitsFinished()) {
asyncCallsPending.incrementAndGet();
- context.callAsync(() -> splitterEnumerator.enumerateSplits(),
this::onSplitsDiscovered);
+ context.callAsync(this::enumerateSplitsWithState,
this::onSplitsDiscovered);
}
signalNoMoreSplitsIfDone();
}
- private void onSplitsDiscovered(List<JdbcSourceSplit> splits, Throwable
error) {
+ /**
+ * Enumerates splits and captures the splitter state that corresponds to
exactly those splits.
+ *
+ * <p>Both are captured on the same worker thread invocation, so the
returned state and splits
+ * belong together. Splits must never be separated from the state that
produced them, otherwise
+ * a checkpoint taken in between would either drop splits or duplicate
them.
+ */
+ private EnumerationResult enumerateSplitsWithState() {
+ final List<JdbcSourceSplit> splits =
splitterEnumerator.enumerateSplits();
+ return new EnumerationResult(splits,
splitterEnumerator.serializableState());
+ }
+
+ private void onSplitsDiscovered(EnumerationResult result, Throwable error)
{
asyncCallsPending.decrementAndGet();
if (error != null) {
LOG.error("Failed to discover splits.", error);
@@ -152,8 +183,15 @@ public class JdbcSourceEnumerator
return;
}
- if (splits != null && !splits.isEmpty()) {
- assignOrBuffer(splits);
+ // The splits are now owned by this enumerator (buffered) or by a
reader, so the state that
+ // produced them may be included in a checkpoint from here on.
Committing it here, on the
+ // coordinator thread, keeps the checkpointed state in sync with the
splits that have
+ // actually been handled; preDiscoverSplits keeps at most one
enumeration in flight so the
+ // states are always committed in the order they were captured.
+ lastHandledSplitterState = result.state;
+
+ if (result.splits != null && !result.splits.isEmpty()) {
+ assignOrBuffer(result.splits);
preDiscoverSplits();
} else if (!splitterEnumerator.isAllSplitsFinished()) {
preDiscoverSplits();
@@ -193,4 +231,19 @@ public class JdbcSourceEnumerator
readersWaitingForSplits.clear();
}
}
+
+ /**
+ * The outcome of a single {@link SplitterEnumerator#enumerateSplits()}
invocation: the produced
+ * splits together with the splitter state as of that invocation.
+ */
+ private static final class EnumerationResult {
+ private final @Nullable List<JdbcSourceSplit> splits;
+ private final @Nullable Serializable state;
+
+ private EnumerationResult(
+ @Nullable List<JdbcSourceSplit> splits, @Nullable Serializable
state) {
+ this.splits = splits;
+ this.state = state;
+ }
+ }
}
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/SqlTemplateSplitEnumerator.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/SqlTemplateSplitEnumerator.java
index f1c6c52f..344d8fd8 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/SqlTemplateSplitEnumerator.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/SqlTemplateSplitEnumerator.java
@@ -146,8 +146,20 @@ public final class SqlTemplateSplitEnumerator extends
JdbcSqlSplitEnumeratorBase
@Override
public JdbcSqlSplitEnumeratorBase<JdbcSourceSplit> create() {
+ // Stateful providers such as JdbcSlideTimingParameterProvider are
mutated in place
+ // while enumerating. Starting from a null state would make a
checkpoint that lands
+ // before the first batch is handled store null; on restore the
provider could then
+ // not be rolled back and the in-flight first batch would be lost.
Falling back to the
+ // provider's current state keeps the enumerator's state non-null
from the start, so
+ // that restore resets the provider to a well-defined point.
+ final Serializable initialState =
+ optionalSqlSplitEnumeratorState != null
+ ? optionalSqlSplitEnumeratorState
+ : (parameterValuesProvider == null
+ ? null
+ :
parameterValuesProvider.getLatestOptionalState());
return new SqlTemplateSplitEnumerator(
- this.optionalSqlSplitEnumeratorState, sqlTemplate,
parameterValuesProvider);
+ initialState, sqlTemplate, parameterValuesProvider);
}
@Override
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumeratorCheckpointRaceTest.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumeratorCheckpointRaceTest.java
new file mode 100644
index 00000000..ef226ca0
--- /dev/null
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumeratorCheckpointRaceTest.java
@@ -0,0 +1,325 @@
+/*
+ * 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.jdbc.core.datastream.source.enumerator;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import
org.apache.flink.connector.jdbc.core.datastream.source.config.ContinuousUnBoundingSettings;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.JdbcSqlSplitterEnumerator;
+import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.SplitterEnumerator;
+import
org.apache.flink.connector.jdbc.core.datastream.source.split.CheckpointedOffset;
+import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
+import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
+import org.apache.flink.connector.jdbc.split.JdbcSlideTimingParameterProvider;
+import
org.apache.flink.connector.testutils.source.reader.TestingSplitEnumeratorContext;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.function.BiConsumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Regression test for FLINK-40779. */
+class JdbcSourceEnumeratorCheckpointRaceTest {
+
+ /**
+ * Reproduces the window that FLINK-40779 describes.
+ *
+ * <p>{@code JdbcSourceEnumerator} enumerates splits on a worker thread
and handles the result
+ * on the coordinator thread. A checkpoint taken in between used to record
the splitter state
+ * that had already advanced past the splits still in flight, so those
splits were lost on
+ * restore.
+ */
+ @Test
+ void testCheckpointBetweenEnumerationAndHandlingDoesNotLoseSplits() throws
Exception {
+ final ControllableAsyncContext context = new
ControllableAsyncContext(1);
+ final CountingSplitterEnumerator splitter = new
CountingSplitterEnumerator();
+ final JdbcSourceEnumerator enumerator =
+ new JdbcSourceEnumerator(context, splitter, null, new
ArrayList<>());
+
+ // Schedules the first enumeration, which has not run yet.
+ enumerator.start();
+
+ // The enumeration runs "on the worker thread": the splitter advances
to state 1 and
+ // produces split-000, but the handler that would buffer split-000 has
not run yet.
+ context.runNextCallable();
+ assertThat(splitter.serializableState()).isEqualTo(1);
+
+ // A checkpoint taken in this window must not claim the advanced
state, because split-000 is
+ // not part of it yet.
+ final JdbcSourceEnumeratorState checkpoint =
enumerator.snapshotState(1L);
+ assertThat(checkpoint.getRemainingSplits()).isEmpty();
+
assertThat(checkpoint.getOptionalUserDefinedSplitEnumeratorState()).isEqualTo(0);
+
+ // Restoring from that checkpoint must produce split-000 again rather
than dropping it.
+ final ControllableAsyncContext restoredContext = new
ControllableAsyncContext(1);
+ final JdbcSourceEnumerator restored =
+ new JdbcSourceEnumerator(
+ restoredContext,
+ splitter.restoreState(
+
checkpoint.getOptionalUserDefinedSplitEnumeratorState()),
+ null,
+ new ArrayList<>(checkpoint.getRemainingSplits()));
+ restored.start();
+ restoredContext.runNextCallable();
+ restoredContext.runNextHandler();
+
+ assertThat(restored.snapshotState(2L).getRemainingSplits())
+ .extracting(JdbcSourceSplit::splitId)
+ .containsExactly("split-000");
+ }
+
+ @Test
+ void testStateIsCommittedOnceSplitsHaveBeenHandled() throws Exception {
+ final ControllableAsyncContext context = new
ControllableAsyncContext(1);
+ final CountingSplitterEnumerator splitter = new
CountingSplitterEnumerator();
+ final JdbcSourceEnumerator enumerator =
+ new JdbcSourceEnumerator(context, splitter, null, new
ArrayList<>());
+
+ enumerator.start();
+ context.runNextCallable();
+ context.runNextHandler();
+
+ // split-000 is now owned by the enumerator, so the state that
produced it is checkpointed
+ // together with it.
+ final JdbcSourceEnumeratorState checkpoint =
enumerator.snapshotState(1L);
+ assertThat(checkpoint.getRemainingSplits())
+ .extracting(JdbcSourceSplit::splitId)
+ .containsExactly("split-000");
+
assertThat(checkpoint.getOptionalUserDefinedSplitEnumeratorState()).isEqualTo(1);
+
+ // Let the next enumeration run on the worker thread: the splitter
advances to 2, but its
+ // handler has not run yet. That state must not leak into the
checkpoint, otherwise the
+ // split it produces would be skipped after a restore. This fails on
the code without the
+ // fix, where snapshotState() reads the live splitter state.
+ context.runNextCallable();
+ assertThat(splitter.serializableState()).isEqualTo(2);
+
+ final JdbcSourceEnumeratorState inFlight =
enumerator.snapshotState(2L);
+ assertThat(inFlight.getRemainingSplits())
+ .extracting(JdbcSourceSplit::splitId)
+ .containsExactly("split-000");
+
assertThat(inFlight.getOptionalUserDefinedSplitEnumeratorState()).isEqualTo(1);
+ }
+
+ @Test
+ void testSetSqlProviderSeedsStateFromProviderWhenNoStateIsConfigured() {
+ final ControllableSlideTimingProvider provider =
+ new ControllableSlideTimingProvider(1000L, 500L, 1L, 0L,
1500L);
+ final SqlTemplateSplitEnumerator.TemplateSqlSplitEnumeratorProvider
providerBuilder =
+ new
SqlTemplateSplitEnumerator.TemplateSqlSplitEnumeratorProvider()
+ .setSqlTemplate("SELECT * FROM t WHERE ts >= ? AND ts
< ?")
+ .setParameterValuesProvider(provider);
+
+
assertThat(providerBuilder.create().enumeratorState()).isEqualTo(1000L);
+ }
+
+ /**
+ * The first batch of a setSql source with a stateful provider must
survive a restore. The
+ * provider is mutated in place while enumerating, so a checkpoint taken
between the enumeration
+ * and its handler used to store a null state that a restore could not
roll back.
+ */
+ @Test
+ void testFirstBatchOfSetSqlSourceIsReEnumeratedAfterRestore() throws
Exception {
+ // now = 1500 produces exactly one window [1000, 1500) and advances
the provider to 1001.
+ final ControllableSlideTimingProvider provider =
+ new ControllableSlideTimingProvider(1000L, 500L, 1L, 0L,
1500L);
+ final JdbcSqlSplitterEnumerator splitter =
+ new JdbcSqlSplitterEnumerator(
+ new
SqlTemplateSplitEnumerator.TemplateSqlSplitEnumeratorProvider()
+ .setSqlTemplate("SELECT * FROM t WHERE ts >= ?
AND ts < ?")
+ .setParameterValuesProvider(provider),
+ new ContinuousUnBoundingSettings(null, Duration.ZERO));
+
+ final ControllableAsyncContext context = new
ControllableAsyncContext(1);
+ final JdbcSourceEnumerator enumerator =
+ new JdbcSourceEnumerator(context, splitter, null, new
ArrayList<>());
+
+ enumerator.start();
+ context.runNextCallable();
+ // The provider advanced on the worker thread, but its split has not
been handled yet.
+ assertThat(provider.getLatestOptionalState()).isEqualTo(1001L);
+
+ final JdbcSourceEnumeratorState checkpoint =
enumerator.snapshotState(1L);
+ // Without the initial-state fallback this would be null and the first
batch would be lost.
+
assertThat(checkpoint.getOptionalUserDefinedSplitEnumeratorState()).isEqualTo(1000L);
+ assertThat(checkpoint.getRemainingSplits()).isEmpty();
+
+ // A global failover reuses the source's splitter, i.e. the already
advanced provider.
+ final ControllableAsyncContext restoredContext = new
ControllableAsyncContext(1);
+ final JdbcSourceEnumerator restored =
+ new JdbcSourceEnumerator(
+ restoredContext,
+ splitter.restoreState(
+
checkpoint.getOptionalUserDefinedSplitEnumeratorState()),
+ null,
+ new ArrayList<>(checkpoint.getRemainingSplits()));
+ restored.start();
+ restoredContext.runNextCallable();
+ restoredContext.runNextHandler();
+
+ assertThat(restored.snapshotState(2L).getRemainingSplits())
+ .extracting(split -> split.getParameters()[0])
+ .containsExactly(1000L);
+ }
+
+ @Test
+ void testAtMostOneEnumerationIsInFlight() throws Exception {
+ final ControllableAsyncContext context = new
ControllableAsyncContext(4);
+ final CountingSplitterEnumerator splitter = new
CountingSplitterEnumerator();
+ final JdbcSourceEnumerator enumerator =
+ new JdbcSourceEnumerator(context, splitter, null, new
ArrayList<>());
+
+ enumerator.start();
+ // Even with an enumerator parallelism of 4, only one enumeration may
be scheduled.
+ assertThat(context.pendingCallableCount()).isEqualTo(1);
+
+ context.runNextCallable();
+ // The handler has not run, so no further enumeration may have been
scheduled.
+ assertThat(context.pendingCallableCount()).isZero();
+ }
+
+ /**
+ * A {@link TestingSplitEnumeratorContext} that keeps the async callable
and its handler apart,
+ * so that a test can run the callable, observe the intermediate state,
and only then run the
+ * handler. The parent class always runs the two back to back, which hides
the race.
+ */
+ private static final class ControllableAsyncContext
+ extends TestingSplitEnumeratorContext<JdbcSourceSplit> {
+
+ private final List<Runnable> pendingCallables = new ArrayList<>();
+ private final List<Runnable> pendingHandlers = new ArrayList<>();
+
+ private ControllableAsyncContext(int parallelism) {
+ super(parallelism);
+ }
+
+ @Override
+ public <T> void callAsync(Callable<T> callable, BiConsumer<T,
Throwable> handler) {
+ pendingCallables.add(
+ () -> {
+ try {
+ final T result = callable.call();
+ pendingHandlers.add(() -> handler.accept(result,
null));
+ } catch (Throwable t) {
+ pendingHandlers.add(() -> handler.accept(null, t));
+ }
+ });
+ }
+
+ private void runNextCallable() {
+ assertThat(pendingCallables).as("a pending callable").isNotEmpty();
+ pendingCallables.remove(0).run();
+ }
+
+ private void runNextHandler() {
+ assertThat(pendingHandlers).as("a pending handler").isNotEmpty();
+ pendingHandlers.remove(0).run();
+ }
+
+ private int pendingCallableCount() {
+ return pendingCallables.size();
+ }
+ }
+
+ /**
+ * A {@link JdbcSlideTimingParameterProvider} with a controllable clock,
so that the number of
+ * produced windows is deterministic and does not depend on wall-clock
time.
+ */
+ private static final class ControllableSlideTimingProvider
+ extends JdbcSlideTimingParameterProvider {
+
+ private final long now;
+
+ private ControllableSlideTimingProvider(
+ long startMillis,
+ long slideSpanMills,
+ long slideStepMills,
+ long splitGenerateDelayMillis,
+ long now) {
+ super(startMillis, slideSpanMills, slideStepMills,
splitGenerateDelayMillis);
+ this.now = now;
+ }
+
+ @Override
+ public Long currentAvailableMillis() {
+ return now;
+ }
+ }
+
+ /**
+ * Produces one split per enumeration and records the number of splits
handed out so far as its
+ * serializable state, mimicking how the real splitter enumerators advance
their cursor.
+ */
+ private static final class CountingSplitterEnumerator implements
SplitterEnumerator {
+
+ private int nextSplit = 0;
+ private int produced = 0;
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.CONTINUOUS_UNBOUNDED;
+ }
+
+ @Override
+ public void start(JdbcConnectionProvider connectionProvider) {}
+
+ @Override
+ public void close() {}
+
+ @Override
+ public boolean isAllSplitsFinished() {
+ return false;
+ }
+
+ @Override
+ public List<JdbcSourceSplit> enumerateSplits() {
+ final String splitId = String.format("split-%03d", nextSplit++);
+ produced = nextSplit;
+ return Collections.singletonList(
+ new JdbcSourceSplit(
+ splitId,
+ "select 1",
+ new Serializable[] {nextSplit},
+ new CheckpointedOffset()));
+ }
+
+ @Override
+ public List<String> lineageQueries() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Serializable serializableState() {
+ return produced;
+ }
+
+ @Override
+ public SplitterEnumerator restoreState(Serializable state) {
+ this.nextSplit = (Integer) state;
+ this.produced = (Integer) state;
+ return this;
+ }
+ }
+}
