This is an automated email from the ASF dual-hosted git repository.
yuxiqian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new e9e77b226b [FLINK-40656][runtime] Reset schema coordinator state on
checkpoint restore (#4541)
e9e77b226b is described below
commit e9e77b226bf2db6d75c9f0dc2302fa3e80845955
Author: Xiaobing Fang <[email protected]>
AuthorDate: Wed Sep 16 15:55:40 2026 +0800
[FLINK-40656][runtime] Reset schema coordinator state on checkpoint restore
(#4541)
---
.../FlinkSchemaCoordinatorRecoveryITCase.java | 152 ++++++
.../testsource/recovery/RecoveryDataSink.java | 107 +++++
.../recovery/RecoveryDataSinkFactory.java | 52 ++
.../testsource/recovery/RecoveryDataSource.java | 48 ++
.../recovery/RecoveryDataSourceFactory.java | 57 +++
.../recovery/RecoverySourceFunction.java | 242 ++++++++++
.../org.apache.flink.cdc.common.factories.Factory | 4 +-
.../flink-cdc-pipeline-e2e-tests/pom.xml | 15 +-
.../tests/SchemaCoordinatorRecoveryE2eITCase.java | 260 ++++++++++
.../tests/recovery/RecoveryE2eFixtureFactory.java | 522 +++++++++++++++++++++
.../tests/utils/PipelineTestEnvironment.java | 8 +-
.../operators/schema/common/SchemaRegistry.java | 110 ++++-
.../schema/distributed/SchemaCoordinator.java | 54 ++-
.../schema/regular/SchemaCoordinator.java | 32 +-
.../schema/distributed/SchemaCoordinatorTest.java | 426 +++++++++++++++++
.../schema/regular/SchemaCoordinatorTest.java | 137 ++++++
16 files changed, 2182 insertions(+), 44 deletions(-)
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkSchemaCoordinatorRecoveryITCase.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkSchemaCoordinatorRecoveryITCase.java
new file mode 100644
index 0000000000..5c5cb7754f
--- /dev/null
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkSchemaCoordinatorRecoveryITCase.java
@@ -0,0 +1,152 @@
+/*
+ * 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.cdc.composer.flink;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.composer.definition.PipelineDef;
+import org.apache.flink.cdc.composer.definition.SinkDef;
+import org.apache.flink.cdc.composer.definition.SourceDef;
+import org.apache.flink.cdc.composer.testsource.recovery.RecoveryDataSink;
+import
org.apache.flink.cdc.composer.testsource.recovery.RecoveryDataSinkFactory;
+import
org.apache.flink.cdc.composer.testsource.recovery.RecoveryDataSourceFactory;
+import
org.apache.flink.cdc.composer.testsource.recovery.RecoverySourceFunction;
+import org.apache.flink.cdc.connectors.values.ValuesDatabase;
+import org.apache.flink.configuration.RestartStrategyOptions;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static
org.apache.flink.configuration.CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests regular and distributed schema coordinator recovery through a real
MiniCluster. */
+@Timeout(value = 60, unit = TimeUnit.SECONDS)
+class FlinkSchemaCoordinatorRecoveryITCase {
+
+ private static final org.apache.flink.configuration.Configuration
MINI_CLUSTER_CONFIG =
+ new org.apache.flink.configuration.Configuration();
+
+ static {
+ MINI_CLUSTER_CONFIG.set(
+ ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL,
+ Collections.singletonList("org.apache.flink.cdc"));
+ }
+
+ @RegisterExtension
+ static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(4)
+ .setConfiguration(MINI_CLUSTER_CONFIG)
+ .build());
+
+ @AfterEach
+ void cleanup() {
+ ValuesDatabase.clear();
+ RecoveryDataSink.reset();
+ RecoverySourceFunction.reset();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void replaysFailedSchemaChangeAfterCheckpointAndContinues(boolean
distributed)
+ throws Exception {
+ RecoveryDataSink.reset();
+ RecoverySourceFunction.reset();
+ ValuesDatabase.clear();
+
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(RecoveryDataSourceFactory.PARALLEL_METADATA_SOURCE,
distributed);
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, distributed ?
2 : 1);
+ pipelineConfig.set(PipelineOptions.PIPELINE_NAME,
"schema-coordinator-recovery");
+
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+ composer.getEnv().enableCheckpointing(50L);
+
composer.getEnv().getCheckpointConfig().setMinPauseBetweenCheckpoints(2_000L);
+ org.apache.flink.configuration.Configuration restartConfig =
+ new org.apache.flink.configuration.Configuration();
+ restartConfig.set(RestartStrategyOptions.RESTART_STRATEGY,
"fixed-delay");
+
restartConfig.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS,
1);
+
restartConfig.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY,
Duration.ZERO);
+ composer.getEnv().configure(restartConfig);
+ composer.compose(
+ new PipelineDef(
+ new SourceDef(
+ RecoveryDataSourceFactory.IDENTIFIER,
+ "Recovery test source",
+ sourceConfig),
+ new SinkDef(
+ RecoveryDataSinkFactory.IDENTIFIER,
+ "Recovery test sink",
+ new Configuration()),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ pipelineConfig));
+
+ JobClient jobClient =
composer.getEnv().executeAsync("schema-coordinator-recovery");
+ CompletableFuture<JobExecutionResult> executionResult =
jobClient.getJobExecutionResult();
+ try {
+ executionResult.get();
+
+ TableId tableId = TableId.tableId("recovery", "schema", "orders");
+ assertThat(RecoveryDataSink.getFailures()).isEqualTo(1);
+
assertThat(RecoverySourceFunction.getRunAttempts()).isGreaterThanOrEqualTo(2);
+
assertThat(RecoverySourceFunction.getFirstColumnEmissions()).isEqualTo(2);
+
assertThat(RecoveryDataSink.getFirstColumnApplications()).isEqualTo(2);
+ assertThat(ValuesDatabase.getTableSchema(tableId).getColumnNames())
+ .containsExactly("id", "value", "extra_v1", "extra_v2");
+ assertThat(ValuesDatabase.getResults(tableId))
+ .containsExactly(
+
"recovery.schema.orders:id=1;value=after;extra_v1=v1;extra_v2=v2");
+ } finally {
+ cancelAndWait(jobClient, executionResult);
+ }
+ }
+
+ private static void cancelAndWait(
+ JobClient jobClient, CompletableFuture<JobExecutionResult>
executionResult)
+ throws Exception {
+ if (!executionResult.isDone()) {
+ jobClient.cancel().get(30, TimeUnit.SECONDS);
+ try {
+ executionResult.get(30, TimeUnit.SECONDS);
+ } catch (CancellationException | ExecutionException ignored) {
+ // A cancelled job completes the result future exceptionally.
+ }
+ }
+ }
+}
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSink.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSink.java
new file mode 100644
index 0000000000..e587e53c47
--- /dev/null
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSink.java
@@ -0,0 +1,107 @@
+/*
+ * 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.cdc.composer.testsource.recovery;
+
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEventType;
+import org.apache.flink.cdc.common.sink.DataSink;
+import org.apache.flink.cdc.common.sink.EventSinkProvider;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.cdc.connectors.values.ValuesDatabase;
+import org.apache.flink.cdc.connectors.values.sink.ValuesDataSink;
+
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/** A values sink whose first metadata application fails after applying the
first added column. */
+public class RecoveryDataSink implements DataSink {
+
+ private static final AtomicBoolean FAILED = new AtomicBoolean();
+ private static final AtomicInteger FAILURES = new AtomicInteger();
+ private static final AtomicInteger FIRST_COLUMN_APPLICATIONS = new
AtomicInteger();
+
+ public static void reset() {
+ FAILED.set(false);
+ FAILURES.set(0);
+ FIRST_COLUMN_APPLICATIONS.set(0);
+ }
+
+ public static int getFailures() {
+ return FAILURES.get();
+ }
+
+ public static int getFirstColumnApplications() {
+ return FIRST_COLUMN_APPLICATIONS.get();
+ }
+
+ @Override
+ public EventSinkProvider getEventSinkProvider() {
+ return new ValuesDataSink(true, false,
ValuesDataSink.SinkApi.SINK_FUNCTION, false)
+ .getEventSinkProvider();
+ }
+
+ @Override
+ public MetadataApplier getMetadataApplier() {
+ return new FailingOnceMetadataApplier();
+ }
+
+ private static class FailingOnceMetadataApplier implements MetadataApplier
{
+
+ private final ValuesDatabase.ValuesMetadataApplier delegate =
+ new ValuesDatabase.ValuesMetadataApplier(true);
+
+ @Override
+ public MetadataApplier setAcceptedSchemaEvolutionTypes(
+ Set<SchemaChangeEventType> schemaEvolutionTypes) {
+ delegate.setAcceptedSchemaEvolutionTypes(schemaEvolutionTypes);
+ return this;
+ }
+
+ @Override
+ public boolean acceptsSchemaEvolutionType(SchemaChangeEventType
schemaChangeEventType) {
+ return delegate.acceptsSchemaEvolutionType(schemaChangeEventType);
+ }
+
+ @Override
+ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
+ return delegate.getSupportedSchemaEvolutionTypes();
+ }
+
+ @Override
+ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) {
+ if (schemaChangeEvent instanceof AddColumnEvent
+ && ((AddColumnEvent) schemaChangeEvent)
+ .getAddedColumns()
+ .get(0)
+ .getAddColumn()
+ .getName()
+ .equals("extra_v1")) {
+ FIRST_COLUMN_APPLICATIONS.incrementAndGet();
+ if (FAILED.compareAndSet(false, true)) {
+ delegate.applySchemaChange(schemaChangeEvent);
+ FAILURES.incrementAndGet();
+ throw new RuntimeException("Fail after applying extra_v1
without rollback.");
+ }
+ return;
+ }
+ delegate.applySchemaChange(schemaChangeEvent);
+ }
+ }
+}
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSinkFactory.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSinkFactory.java
new file mode 100644
index 0000000000..633eae6ca4
--- /dev/null
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSinkFactory.java
@@ -0,0 +1,52 @@
+/*
+ * 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.cdc.composer.testsource.recovery;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.factories.DataSinkFactory;
+import org.apache.flink.cdc.common.sink.DataSink;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableSet;
+
+import java.util.Set;
+
+/** Factory for {@link RecoveryDataSink}. */
+public class RecoveryDataSinkFactory implements DataSinkFactory {
+
+ public static final String IDENTIFIER = "recovery-test-sink";
+
+ @Override
+ public DataSink createDataSink(Context context) {
+ return new RecoveryDataSink();
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return ImmutableSet.of();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return ImmutableSet.of();
+ }
+}
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSource.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSource.java
new file mode 100644
index 0000000000..a2bedb5917
--- /dev/null
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSource.java
@@ -0,0 +1,48 @@
+/*
+ * 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.cdc.composer.testsource.recovery;
+
+import org.apache.flink.cdc.common.source.DataSource;
+import org.apache.flink.cdc.common.source.EventSourceProvider;
+import org.apache.flink.cdc.common.source.FlinkSourceFunctionProvider;
+import org.apache.flink.cdc.common.source.MetadataAccessor;
+
+/** A checkpoint-aware source used by schema coordinator recovery integration
tests. */
+public class RecoveryDataSource implements DataSource {
+
+ private final boolean parallelMetadataSource;
+
+ public RecoveryDataSource(boolean parallelMetadataSource) {
+ this.parallelMetadataSource = parallelMetadataSource;
+ }
+
+ @Override
+ public EventSourceProvider getEventSourceProvider() {
+ return FlinkSourceFunctionProvider.of(new RecoverySourceFunction());
+ }
+
+ @Override
+ public MetadataAccessor getMetadataAccessor() {
+ throw new UnsupportedOperationException("Recovery test source has no
metadata accessor.");
+ }
+
+ @Override
+ public boolean isParallelMetadataSource() {
+ return parallelMetadataSource;
+ }
+}
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSourceFactory.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSourceFactory.java
new file mode 100644
index 0000000000..120abdd425
--- /dev/null
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoveryDataSourceFactory.java
@@ -0,0 +1,57 @@
+/*
+ * 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.cdc.composer.testsource.recovery;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.configuration.ConfigOptions;
+import org.apache.flink.cdc.common.factories.DataSourceFactory;
+import org.apache.flink.cdc.common.source.DataSource;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableSet;
+
+import java.util.Set;
+
+/** Factory for {@link RecoveryDataSource}. */
+public class RecoveryDataSourceFactory implements DataSourceFactory {
+
+ public static final String IDENTIFIER = "recovery-test-source";
+
+ public static final ConfigOption<Boolean> PARALLEL_METADATA_SOURCE =
+
ConfigOptions.key("parallel-metadata-source").booleanType().defaultValue(false);
+
+ @Override
+ public DataSource createDataSource(Context context) {
+ return new RecoveryDataSource(
+
context.getFactoryConfiguration().get(PARALLEL_METADATA_SOURCE));
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return ImmutableSet.of();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return ImmutableSet.of(PARALLEL_METADATA_SOURCE);
+ }
+}
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoverySourceFunction.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoverySourceFunction.java
new file mode 100644
index 0000000000..8877a9e1ef
--- /dev/null
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsource/recovery/RecoverySourceFunction.java
@@ -0,0 +1,242 @@
+/*
+ * 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.cdc.composer.testsource.recovery;
+
+import org.apache.flink.api.common.state.CheckpointListener;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.cdc.source.RuntimeContextAdapter;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import
org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/** Emits an add-column event only after a completed checkpoint and restores
its event cursor. */
+public class RecoverySourceFunction extends RichParallelSourceFunction<Event>
+ implements CheckpointedFunction, CheckpointListener {
+
+ private static final int ADD_FIRST_COLUMN_POSITION = 2;
+
+ private static final AtomicInteger RUN_ATTEMPTS = new AtomicInteger();
+ private static final AtomicInteger FIRST_COLUMN_EMISSIONS = new
AtomicInteger();
+
+ private transient ListState<Integer> cursorState;
+ private transient Map<Long, Integer> checkpointCursors;
+ private volatile boolean running = true;
+ private volatile boolean checkpointCompleted;
+ private int cursor;
+ private boolean restored;
+
+ public static void reset() {
+ RUN_ATTEMPTS.set(0);
+ FIRST_COLUMN_EMISSIONS.set(0);
+ }
+
+ public static int getRunAttempts() {
+ return RUN_ATTEMPTS.get();
+ }
+
+ public static int getFirstColumnEmissions() {
+ return FIRST_COLUMN_EMISSIONS.get();
+ }
+
+ @Override
+ public void run(SourceFunction.SourceContext<Event> context) throws
Exception {
+ if (RuntimeContextAdapter.getIndexOfThisSubtask(
+ (StreamingRuntimeContext) getRuntimeContext())
+ != 0) {
+ return;
+ }
+ RUN_ATTEMPTS.incrementAndGet();
+ List<Event> events = events();
+ if (restored) {
+ synchronized (context.getCheckpointLock()) {
+ context.collect(events.get(0));
+ }
+ }
+ while (running && cursor < events.size()) {
+ if (cursor == ADD_FIRST_COLUMN_POSITION && !checkpointCompleted) {
+ Thread.sleep(10L);
+ continue;
+ }
+ if (cursor == ADD_FIRST_COLUMN_POSITION) {
+ synchronized (context.getCheckpointLock()) {
+ FIRST_COLUMN_EMISSIONS.incrementAndGet();
+ context.collect(events.get(cursor));
+ if (restored) {
+ cursor++;
+ }
+ }
+ // Preserve cursor 2 until the first attempt is cancelled, so
it must replay C.
+ while (running && !restored) {
+ Thread.sleep(10L);
+ }
+ continue;
+ }
+ synchronized (context.getCheckpointLock()) {
+ context.collect(events.get(cursor));
+ cursor++;
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ running = false;
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws
Exception {
+ cursorState.update(Collections.singletonList(cursor));
+ checkpointCursors.put(context.getCheckpointId(), cursor);
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws
Exception {
+ cursorState =
+ context.getOperatorStateStore()
+ .getListState(new
ListStateDescriptor<>("event-cursor", Types.INT));
+ checkpointCursors = new HashMap<>();
+ for (Integer restoredCursor : cursorState.get()) {
+ cursor = restoredCursor;
+ }
+ restored = context.isRestored();
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {
+ Integer checkpointCursor = checkpointCursors.remove(checkpointId);
+ if (checkpointCursor != null && checkpointCursor ==
ADD_FIRST_COLUMN_POSITION) {
+ checkpointCompleted = true;
+ }
+ }
+
+ private static List<Event> events() {
+ TableId tableId = TableId.tableId("recovery", "schema", "orders");
+ Schema initialSchema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.STRING())
+ .physicalColumn("value", DataTypes.STRING())
+ .primaryKey("id")
+ .build();
+ BinaryRecordDataGenerator twoColumns =
+ new BinaryRecordDataGenerator(
+ new DataType[] {DataTypes.STRING(),
DataTypes.STRING()});
+ BinaryRecordDataGenerator threeColumns =
+ new BinaryRecordDataGenerator(
+ new DataType[] {
+ DataTypes.STRING(), DataTypes.STRING(),
DataTypes.STRING()
+ });
+ BinaryRecordDataGenerator fourColumns =
+ new BinaryRecordDataGenerator(
+ new DataType[] {
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ });
+ return Arrays.asList(
+ new CreateTableEvent(tableId, initialSchema),
+ DataChangeEvent.insertEvent(
+ tableId,
+ twoColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("before")
+ })),
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("extra_v1",
DataTypes.STRING())))),
+ DataChangeEvent.updateEvent(
+ tableId,
+ threeColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("before"),
+ null
+ }),
+ threeColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("after"),
+ BinaryStringData.fromString("v1")
+ })),
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("extra_v2",
DataTypes.STRING())))),
+ DataChangeEvent.updateEvent(
+ tableId,
+ fourColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("after"),
+ BinaryStringData.fromString("v1"),
+ null
+ }),
+ fourColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("after"),
+ BinaryStringData.fromString("v1"),
+ BinaryStringData.fromString("v2")
+ })),
+ DataChangeEvent.insertEvent(
+ tableId,
+ fourColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("deleted"),
+ BinaryStringData.fromString("v1"),
+ BinaryStringData.fromString("v2")
+ })),
+ DataChangeEvent.deleteEvent(
+ tableId,
+ fourColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("deleted"),
+ BinaryStringData.fromString("v1"),
+ BinaryStringData.fromString("v2")
+ })));
+ }
+}
diff --git
a/flink-cdc-composer/src/test/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
b/flink-cdc-composer/src/test/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
index 274faa0078..f5be4c8416 100644
---
a/flink-cdc-composer/src/test/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
+++
b/flink-cdc-composer/src/test/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
@@ -17,5 +17,7 @@ org.apache.flink.cdc.composer.utils.factory.DataSinkFactory2
org.apache.flink.cdc.composer.utils.factory.DataSourceFactory1
org.apache.flink.cdc.composer.utils.factory.DataSourceFactory2
org.apache.flink.cdc.composer.testsource.factory.DistributedDataSourceFactory
+org.apache.flink.cdc.composer.testsource.recovery.RecoveryDataSourceFactory
+org.apache.flink.cdc.composer.testsource.recovery.RecoveryDataSinkFactory
org.apache.flink.cdc.composer.flink.FlinkPipelineComposerTest$TestDataSinkFactory
-org.apache.flink.cdc.composer.flink.FlinkPipelineComposerTest$TestDataSourceFactory
\ No newline at end of file
+org.apache.flink.cdc.composer.flink.FlinkPipelineComposerTest$TestDataSourceFactory
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
index 37aec13dcf..7e33a006d3 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
@@ -36,6 +36,7 @@ limitations under the License.
<flink.release.download.skip>false</flink.release.download.skip>
<flink.release.name>flink-${flink.version}-bin-scala_${scala.binary.version}.tgz</flink.release.name>
<flink.release.mirror>https://dlcdn.apache.org/flink/flink-${flink.version}</flink.release.mirror>
+
<flink.hadoop.compatibility.artifact-id>flink-hadoop-compatibility_2.12</flink.hadoop.compatibility.artifact-id>
<maven.plugin.download.version>1.6.8</maven.plugin.download.version>
<iceberg.version>1.10.1</iceberg.version>
<hive.version>2.3.9</hive.version>
@@ -169,7 +170,7 @@ limitations under the License.
</dependency>
<dependency>
<groupId>org.apache.paimon</groupId>
- <artifactId>paimon-flink-${flink.major.version}</artifactId>
+ <artifactId>paimon-flink-${paimon.flink.major.version}</artifactId>
<version>${paimon.version}</version>
<scope>test</scope>
</dependency>
@@ -775,7 +776,7 @@ limitations under the License.
</artifactItem>
<artifactItem>
<groupId>org.apache.flink</groupId>
-
<artifactId>flink-hadoop-compatibility_2.12</artifactId>
+
<artifactId>${flink.hadoop.compatibility.artifact-id}</artifactId>
<version>${flink.version}</version>
<destFileName>flink-hadoop-compatibility.jar</destFileName>
<type>jar</type>
@@ -843,4 +844,14 @@ limitations under the License.
</plugin>
</plugins>
</build>
+
+ <profiles>
+ <profile>
+ <id>flink2</id>
+ <properties>
+ <flink.version>${flink.2.x.version}</flink.version>
+
<flink.hadoop.compatibility.artifact-id>flink-hadoop-compatibility</flink.hadoop.compatibility.artifact-id>
+ </properties>
+ </profile>
+ </profiles>
</project>
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaCoordinatorRecoveryE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaCoordinatorRecoveryE2eITCase.java
new file mode 100644
index 0000000000..d70f85386c
--- /dev/null
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaCoordinatorRecoveryE2eITCase.java
@@ -0,0 +1,260 @@
+/*
+ * 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.cdc.pipeline.tests;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.cdc.pipeline.tests.recovery.RecoveryE2eFixtureFactory;
+import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.testcontainers.containers.Container.ExecResult;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.jar.JarEntry;
+import java.util.jar.JarOutputStream;
+import java.util.stream.Collectors;
+
+/** End-to-end recovery coverage for regular and distributed schema
coordinators. */
+class SchemaCoordinatorRecoveryE2eITCase extends PipelineTestEnvironment {
+
+ private static final Duration JOB_TIMEOUT = Duration.ofMinutes(2);
+
+ @TempDir Path tempDir;
+
+ @Override
+ protected String getFlinkProperties() {
+ return super.getFlinkProperties()
+ .replace(
+ "restart-strategy.type: off",
+ "restart-strategy.type: fixed-delay\n"
+ + "restart-strategy.fixed-delay.attempts: 1\n"
+ + "restart-strategy.fixed-delay.delay: 0 s\n"
+ + "execution.checkpointing.min-pause: 2 s");
+ }
+
+ @ParameterizedTest(name = "parallel metadata source: {0}")
+ @ValueSource(booleans = {false, true})
+ void recoversFailedSchemaChangeAndContinues(boolean distributed) throws
Exception {
+ String evidencePath = "/tmp/shared/schema-coordinator-recovery-" +
UUID.randomUUID();
+ Path fixtureJar = createFixtureJar();
+ String pipelineJob =
+ String.format(
+ "source:\n"
+ + " type: %s\n"
+ + " evidence-path: %s\n"
+ + " parallel-metadata-source: %s\n"
+ + "\n"
+ + "sink:\n"
+ + " type: %s\n"
+ + " evidence-path: %s\n"
+ + "\n"
+ + "pipeline:\n"
+ + " parallelism: %d\n"
+ + " schema.change.behavior: lenient\n",
+ RecoveryE2eFixtureFactory.IDENTIFIER,
+ evidencePath,
+ distributed,
+ RecoveryE2eFixtureFactory.IDENTIFIER,
+ evidencePath,
+ distributed ? 2 : 1);
+
+ JobID jobId = submitPipelineJob(pipelineJob, fixtureJar);
+ Assertions.assertThat(jobId).as("pipeline job ID").isNotNull();
+ waitUntilJobFinished(JOB_TIMEOUT);
+
Assertions.assertThat(getRestClusterClient().getJobStatus(jobId).get(10,
TimeUnit.SECONDS))
+ .as("pipeline job completion")
+ .isEqualTo(JobStatus.FINISHED);
+
+ List<String> evidence =
+ Arrays.asList(readContainerFile(evidencePath +
".events").split("\\n"));
+ Assertions.assertThat(occurrences(evidence, "SOURCE\tattempt"))
+ .as("job was restarted after the metadata failure")
+ .isGreaterThanOrEqualTo(2);
+ Assertions.assertThat(occurrences(evidence,
"SOURCE\textra_v1-emitted"))
+ .as("the failed schema event was replayed")
+ .isEqualTo(2);
+ assertRestoredFromCompletedCheckpoint(evidence);
+ Assertions.assertThat(evidence)
+ .contains(
+ "SOURCE\tcreate-replayed",
+ "METADATA\textra_v1-applied",
+ "METADATA\textra_v1-replayed",
+ "METADATA\textra_v2-applied");
+ Assertions.assertThat(occurrences(evidence,
"METADATA\textra_v1-applied"))
+ .as("the no-rollback metadata effect happened once")
+ .isEqualTo(1);
+ Assertions.assertThat(occurrences(evidence,
"METADATA\textra_v1-replayed"))
+ .as("the metadata failure was replayed once")
+ .isEqualTo(1);
+ Assertions.assertThat(occurrences(evidence,
"METADATA\textra_v2-applied"))
+ .as("the follow-up schema change was applied once")
+ .isEqualTo(1);
+ String schema = readContainerFile(evidencePath + ".schema").trim();
+ Assertions.assertThat(schema)
+ .as("follow-up schema change completed after recovery")
+ .isEqualTo("id,value,extra_v1,extra_v2");
+ Assertions.assertThat(dataEvents(evidence))
+ .as("complete data-change multiset")
+ .containsExactlyInAnyOrder(
+ "DATA\tINSERT\t\t=>\t1,before",
+ "DATA\tUPDATE\t1,before,\t=>\t1,after,v1",
+ "DATA\tUPDATE\t1,after,v1,\t=>\t1,after,v1,v2",
+ "DATA\tINSERT\t\t=>\t2,deleted,v1,v2",
+ "DATA\tDELETE\t2,deleted,v1,v2\t=>\t");
+ Assertions.assertThat(terminalRows(evidence, schema))
+ .as("PK upsert and deletion materialized from every emitted
data event")
+ .containsExactly(Map.entry("1", "1,after,v1,v2"));
+ }
+
+ private Path createFixtureJar() throws IOException {
+ Path fixtureJar =
tempDir.resolve("schema-coordinator-recovery-fixture.jar");
+ try (JarOutputStream output = new
JarOutputStream(Files.newOutputStream(fixtureJar))) {
+ for (Class<?> fixtureClass :
RecoveryE2eFixtureFactory.fixtureClasses()) {
+ addClassToJar(fixtureClass, output);
+ }
+ output.putNextEntry(
+ new JarEntry(
+
"META-INF/services/org.apache.flink.cdc.common.factories.Factory"));
+ output.write(
+ (RecoveryE2eFixtureFactory.class.getName() + "\n")
+ .getBytes(StandardCharsets.UTF_8));
+ output.closeEntry();
+ }
+ return fixtureJar;
+ }
+
+ private static void addClassToJar(Class<?> clazz, JarOutputStream output)
throws IOException {
+ String resourceName = clazz.getName().replace('.', '/') + ".class";
+ try (InputStream input =
clazz.getClassLoader().getResourceAsStream(resourceName)) {
+ Assertions.assertThat(input).as("class resource %s",
resourceName).isNotNull();
+ output.putNextEntry(new JarEntry(resourceName));
+ input.transferTo(output);
+ output.closeEntry();
+ }
+ }
+
+ private String readContainerFile(String path) throws IOException,
InterruptedException {
+ ExecResult result = jobManager.execInContainer("cat", path);
+ Assertions.assertThat(result.getExitCode()).as("read evidence file
%s", path).isZero();
+ return result.getStdout();
+ }
+
+ private static int occurrences(List<String> evidence, String expected) {
+ return (int) evidence.stream().filter(expected::equals).count();
+ }
+
+ private static List<String> dataEvents(List<String> evidence) {
+ return evidence.stream()
+ .filter(line -> line.startsWith("DATA\t"))
+ .collect(Collectors.toList());
+ }
+
+ private static void assertRestoredFromCompletedCheckpoint(List<String>
evidence) {
+ int firstColumnEmission = evidence.indexOf("SOURCE\textra_v1-emitted");
+ int restoredCheckpoint = indexOfPrefix(evidence,
"SOURCE\tcheckpoint-restored\t");
+ Assertions.assertThat(firstColumnEmission).as("initial extra_v1
emission").isGreaterThan(0);
+ Assertions.assertThat(restoredCheckpoint)
+ .as("source restore after the failure")
+ .isGreaterThan(firstColumnEmission);
+
+ int completedCheckpointIndex = -1;
+ for (int index = 0; index < firstColumnEmission; index++) {
+ if
(evidence.get(index).startsWith("SOURCE\tcheckpoint-completed\t")) {
+ completedCheckpointIndex = index;
+ }
+ }
+ Assertions.assertThat(completedCheckpointIndex)
+ .as("checkpoint completed before initial extra_v1 emission")
+ .isGreaterThanOrEqualTo(0);
+
+ String completedCheckpoint = evidence.get(completedCheckpointIndex);
+ String checkpointId = completedCheckpoint.split("\\t", -1)[2];
+ String snapshot = "SOURCE\tcheckpoint-snapshot\t" + checkpointId +
"\tcursor=2";
+ int snapshotIndex = evidence.indexOf(snapshot);
+ Assertions.assertThat(snapshotIndex)
+ .as("selected checkpoint captured the pre-failure cursor")
+ .isGreaterThanOrEqualTo(0);
+ Assertions.assertThat(snapshotIndex)
+ .as("checkpoint snapshot precedes its completion")
+ .isLessThan(completedCheckpointIndex);
+ Assertions.assertThat(evidence)
+ .contains(snapshot, "SOURCE\tcheckpoint-restored\t" +
checkpointId + "\tcursor=2");
+ for (int index = snapshotIndex; index < restoredCheckpoint; index++) {
+ String line = evidence.get(index);
+ if (line.startsWith("SOURCE\tcheckpoint-snapshot\t")
+ || line.startsWith("SOURCE\tcheckpoint-completed\t")) {
+ Assertions.assertThat(line)
+ .as("no checkpoint may advance past the replay cursor
before restore")
+ .endsWith("\tcursor=2");
+ }
+ }
+ }
+
+ private static int indexOfPrefix(List<String> evidence, String prefix) {
+ for (int index = 0; index < evidence.size(); index++) {
+ if (evidence.get(index).startsWith(prefix)) {
+ return index;
+ }
+ }
+ return -1;
+ }
+
+ private static Map<String, String> terminalRows(List<String> evidence,
String schema) {
+ Map<String, String> rows = new LinkedHashMap<>();
+ for (String line : evidence) {
+ if (!line.startsWith("DATA\t")) {
+ continue;
+ }
+ String[] parts = line.split("\\t", -1);
+ String operation = parts[1];
+ String row = "DELETE".equals(operation) ? parts[2] : parts[4];
+ String id = row.split(",", -1)[0];
+ if ("DELETE".equals(operation)) {
+ rows.remove(id);
+ } else {
+ rows.put(id, row);
+ }
+ }
+ int columnCount = schema.split(",", -1).length;
+ rows.replaceAll((id, row) -> padToColumnCount(row, columnCount));
+ return rows;
+ }
+
+ private static String padToColumnCount(String row, int columnCount) {
+ StringBuilder paddedRow = new StringBuilder(row);
+ for (int index = row.split(",", -1).length; index < columnCount;
index++) {
+ paddedRow.append(',');
+ }
+ return paddedRow.toString();
+ }
+}
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/recovery/RecoveryE2eFixtureFactory.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/recovery/RecoveryE2eFixtureFactory.java
new file mode 100644
index 0000000000..d34af765f6
--- /dev/null
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/recovery/RecoveryE2eFixtureFactory.java
@@ -0,0 +1,522 @@
+/*
+ * 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.cdc.pipeline.tests.recovery;
+
+import org.apache.flink.api.common.state.CheckpointListener;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.configuration.ConfigOptions;
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.factories.DataSinkFactory;
+import org.apache.flink.cdc.common.factories.DataSourceFactory;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.sink.DataSink;
+import org.apache.flink.cdc.common.sink.EventSinkProvider;
+import org.apache.flink.cdc.common.sink.FlinkSinkFunctionProvider;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.cdc.common.source.DataSource;
+import org.apache.flink.cdc.common.source.EventSourceProvider;
+import org.apache.flink.cdc.common.source.FlinkSourceFunctionProvider;
+import org.apache.flink.cdc.common.source.MetadataAccessor;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.cdc.source.RuntimeContextAdapter;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import
org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Test-only source and sink fixture packaged into the pipeline job JAR. */
+public class RecoveryE2eFixtureFactory implements DataSourceFactory,
DataSinkFactory {
+
+ public static final String IDENTIFIER = "recovery-e2e";
+
+ public static final ConfigOption<String> EVIDENCE_PATH =
+ ConfigOptions.key("evidence-path").stringType().noDefaultValue();
+
+ public static final ConfigOption<Boolean> PARALLEL_METADATA_SOURCE =
+
ConfigOptions.key("parallel-metadata-source").booleanType().defaultValue(false);
+
+ @Override
+ public DataSource createDataSource(DataSourceFactory.Context context) {
+ return new RecoveryE2eDataSource(
+ context.getFactoryConfiguration().get(EVIDENCE_PATH),
+
context.getFactoryConfiguration().get(PARALLEL_METADATA_SOURCE));
+ }
+
+ @Override
+ public DataSink createDataSink(DataSinkFactory.Context context) {
+ return new
RecoveryE2eDataSink(context.getFactoryConfiguration().get(EVIDENCE_PATH));
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Collections.singleton(EVIDENCE_PATH);
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Collections.singleton(PARALLEL_METADATA_SOURCE);
+ }
+
+ public static Class<?>[] fixtureClasses() {
+ return new Class<?>[] {
+ RecoveryE2eFixtureFactory.class,
+ RecoveryE2eDataSource.class,
+ RecoveryE2eDataSink.class,
+ RecoveryE2eSourceFunction.class,
+ RecoveryE2eSinkFunction.class,
+ RecoveryE2eMetadataApplier.class,
+ RecoveryE2eEvents.class,
+ RecoveryE2eEvidence.class
+ };
+ }
+}
+
+class RecoveryE2eDataSource implements DataSource {
+
+ private final String evidencePath;
+ private final boolean parallelMetadataSource;
+
+ RecoveryE2eDataSource(String evidencePath, boolean parallelMetadataSource)
{
+ this.evidencePath = evidencePath;
+ this.parallelMetadataSource = parallelMetadataSource;
+ }
+
+ @Override
+ public EventSourceProvider getEventSourceProvider() {
+ return FlinkSourceFunctionProvider.of(new
RecoveryE2eSourceFunction(evidencePath));
+ }
+
+ @Override
+ public MetadataAccessor getMetadataAccessor() {
+ throw new UnsupportedOperationException("Recovery E2E source has no
metadata accessor.");
+ }
+
+ @Override
+ public boolean isParallelMetadataSource() {
+ return parallelMetadataSource;
+ }
+}
+
+class RecoveryE2eDataSink implements DataSink {
+
+ private final String evidencePath;
+
+ RecoveryE2eDataSink(String evidencePath) {
+ this.evidencePath = evidencePath;
+ }
+
+ @Override
+ public EventSinkProvider getEventSinkProvider() {
+ return FlinkSinkFunctionProvider.of(new
RecoveryE2eSinkFunction(evidencePath));
+ }
+
+ @Override
+ public MetadataApplier getMetadataApplier() {
+ return new RecoveryE2eMetadataApplier(evidencePath);
+ }
+}
+
+class RecoveryE2eSourceFunction extends RichParallelSourceFunction<Event>
+ implements CheckpointedFunction, CheckpointListener {
+
+ private static final int ADD_FIRST_COLUMN_POSITION = 2;
+
+ private final String evidencePath;
+ private transient ListState<Integer> cursorState;
+ private transient ListState<Long> checkpointIdState;
+ private transient Map<Long, Integer> checkpointCursors;
+ private volatile boolean running = true;
+ private volatile boolean checkpointCompleted;
+ private int cursor;
+ private long restoredCheckpointId = -1L;
+ private boolean restored;
+
+ RecoveryE2eSourceFunction(String evidencePath) {
+ this.evidencePath = evidencePath;
+ }
+
+ @Override
+ public void run(SourceFunction.SourceContext<Event> context) throws
Exception {
+ if (RuntimeContextAdapter.getIndexOfThisSubtask(
+ (StreamingRuntimeContext) getRuntimeContext())
+ != 0) {
+ return;
+ }
+
+ RecoveryE2eEvidence.append(evidencePath, "SOURCE\tattempt");
+ List<Event> events = RecoveryE2eEvents.events();
+ if (restored) {
+ synchronized (context.getCheckpointLock()) {
+ context.collect(events.get(0));
+ }
+ RecoveryE2eEvidence.append(evidencePath,
"SOURCE\tcreate-replayed");
+ }
+
+ while (running && cursor < events.size()) {
+ if (cursor == ADD_FIRST_COLUMN_POSITION && !checkpointCompleted) {
+ Thread.sleep(10L);
+ continue;
+ }
+ if (cursor == ADD_FIRST_COLUMN_POSITION) {
+ synchronized (context.getCheckpointLock()) {
+ RecoveryE2eEvidence.append(evidencePath,
"SOURCE\textra_v1-emitted");
+ context.collect(events.get(cursor));
+ if (restored) {
+ cursor++;
+ }
+ }
+ while (running && !restored) {
+ Thread.sleep(10L);
+ }
+ continue;
+ }
+ synchronized (context.getCheckpointLock()) {
+ context.collect(events.get(cursor));
+ cursor++;
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ running = false;
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws
Exception {
+ cursorState.update(Collections.singletonList(cursor));
+
checkpointIdState.update(Collections.singletonList(context.getCheckpointId()));
+ checkpointCursors.put(context.getCheckpointId(), cursor);
+ if (isEmitterSubtask()) {
+ RecoveryE2eEvidence.append(
+ evidencePath,
+ "SOURCE\tcheckpoint-snapshot\t"
+ + context.getCheckpointId()
+ + "\tcursor="
+ + cursor);
+ }
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws
Exception {
+ cursorState =
+ context.getOperatorStateStore()
+ .getListState(new
ListStateDescriptor<>("event-cursor", Types.INT));
+ checkpointIdState =
+ context.getOperatorStateStore()
+ .getListState(new
ListStateDescriptor<>("checkpoint-id", Types.LONG));
+ checkpointCursors = new HashMap<>();
+ for (Integer restoredCursor : cursorState.get()) {
+ cursor = restoredCursor;
+ }
+ for (Long checkpointId : checkpointIdState.get()) {
+ restoredCheckpointId = checkpointId;
+ }
+ restored = context.isRestored();
+ if (restored && isEmitterSubtask()) {
+ RecoveryE2eEvidence.append(
+ evidencePath,
+ "SOURCE\tcheckpoint-restored\t" + restoredCheckpointId +
"\tcursor=" + cursor);
+ }
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {
+ Integer checkpointCursor = checkpointCursors.remove(checkpointId);
+ if (checkpointCursor == null || checkpointCursor !=
ADD_FIRST_COLUMN_POSITION) {
+ return;
+ }
+ if (isEmitterSubtask()) {
+ RecoveryE2eEvidence.append(
+ evidencePath,
+ "SOURCE\tcheckpoint-completed\t"
+ + checkpointId
+ + "\tcursor="
+ + checkpointCursor);
+ }
+ checkpointCompleted = true;
+ }
+
+ private boolean isEmitterSubtask() {
+ return RuntimeContextAdapter.getIndexOfThisSubtask(
+ (StreamingRuntimeContext) getRuntimeContext())
+ == 0;
+ }
+}
+
+class RecoveryE2eSinkFunction implements SinkFunction<Event> {
+
+ private final String evidencePath;
+
+ RecoveryE2eSinkFunction(String evidencePath) {
+ this.evidencePath = evidencePath;
+ }
+
+ @Override
+ public void invoke(Event event, Context context) {
+ if (event instanceof DataChangeEvent) {
+ DataChangeEvent dataChangeEvent = (DataChangeEvent) event;
+ RecoveryE2eEvidence.append(
+ evidencePath,
+ "DATA\t"
+ + dataChangeEvent.op()
+ + "\t"
+ + format(dataChangeEvent.before())
+ + "\t=>\t"
+ + format(dataChangeEvent.after()));
+ }
+ }
+
+ private static String format(RecordData record) {
+ if (record == null) {
+ return "";
+ }
+ StringBuilder builder = new StringBuilder();
+ for (int index = 0; index < record.getArity(); index++) {
+ if (index > 0) {
+ builder.append(',');
+ }
+ if (!record.isNullAt(index)) {
+ builder.append(record.getString(index));
+ }
+ }
+ return builder.toString();
+ }
+}
+
+class RecoveryE2eMetadataApplier implements MetadataApplier {
+
+ private final String evidencePath;
+
+ RecoveryE2eMetadataApplier(String evidencePath) {
+ this.evidencePath = evidencePath;
+ }
+
+ @Override
+ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) {
+ if (schemaChangeEvent instanceof CreateTableEvent) {
+ RecoveryE2eEvidence.initializeSchema(evidencePath, "id,value");
+ RecoveryE2eEvidence.append(evidencePath, "METADATA\tcreate");
+ return;
+ }
+ if (!(schemaChangeEvent instanceof AddColumnEvent)) {
+ return;
+ }
+
+ String columnName =
+ ((AddColumnEvent) schemaChangeEvent)
+ .getAddedColumns()
+ .get(0)
+ .getAddColumn()
+ .getName();
+ if ("extra_v1".equals(columnName)) {
+ if (RecoveryE2eEvidence.createMarker(evidencePath +
".extra-v1-applied")) {
+ RecoveryE2eEvidence.writeSchema(evidencePath,
"id,value,extra_v1");
+ RecoveryE2eEvidence.append(evidencePath,
"METADATA\textra_v1-applied");
+ throw new RuntimeException("Fail after applying extra_v1
without rollback.");
+ }
+ RecoveryE2eEvidence.append(evidencePath,
"METADATA\textra_v1-replayed");
+ return;
+ }
+ if ("extra_v2".equals(columnName)) {
+ RecoveryE2eEvidence.writeSchema(evidencePath,
"id,value,extra_v1,extra_v2");
+ RecoveryE2eEvidence.append(evidencePath,
"METADATA\textra_v2-applied");
+ }
+ }
+}
+
+class RecoveryE2eEvents {
+
+ private RecoveryE2eEvents() {}
+
+ static List<Event> events() {
+ TableId tableId = TableId.tableId("recovery", "schema", "orders");
+ Schema initialSchema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.STRING())
+ .physicalColumn("value", DataTypes.STRING())
+ .primaryKey("id")
+ .build();
+ BinaryRecordDataGenerator twoColumns =
+ new BinaryRecordDataGenerator(
+ new DataType[] {DataTypes.STRING(),
DataTypes.STRING()});
+ BinaryRecordDataGenerator threeColumns =
+ new BinaryRecordDataGenerator(
+ new DataType[] {
+ DataTypes.STRING(), DataTypes.STRING(),
DataTypes.STRING()
+ });
+ BinaryRecordDataGenerator fourColumns =
+ new BinaryRecordDataGenerator(
+ new DataType[] {
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ });
+ return Arrays.asList(
+ new CreateTableEvent(tableId, initialSchema),
+ DataChangeEvent.insertEvent(
+ tableId,
+ twoColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("before")
+ })),
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("extra_v1",
DataTypes.STRING())))),
+ DataChangeEvent.updateEvent(
+ tableId,
+ threeColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("before"),
+ null
+ }),
+ threeColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("after"),
+ BinaryStringData.fromString("v1")
+ })),
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("extra_v2",
DataTypes.STRING())))),
+ DataChangeEvent.updateEvent(
+ tableId,
+ fourColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("after"),
+ BinaryStringData.fromString("v1"),
+ null
+ }),
+ fourColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("after"),
+ BinaryStringData.fromString("v1"),
+ BinaryStringData.fromString("v2")
+ })),
+ DataChangeEvent.insertEvent(
+ tableId,
+ fourColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("deleted"),
+ BinaryStringData.fromString("v1"),
+ BinaryStringData.fromString("v2")
+ })),
+ DataChangeEvent.deleteEvent(
+ tableId,
+ fourColumns.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("deleted"),
+ BinaryStringData.fromString("v1"),
+ BinaryStringData.fromString("v2")
+ })));
+ }
+}
+
+class RecoveryE2eEvidence {
+
+ private RecoveryE2eEvidence() {}
+
+ static void append(String evidencePath, String line) {
+ try {
+ Files.write(
+ Paths.get(evidencePath + ".events"),
+ (line +
System.lineSeparator()).getBytes(StandardCharsets.UTF_8),
+ StandardOpenOption.CREATE,
+ StandardOpenOption.APPEND);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ static boolean createMarker(String markerPath) {
+ try {
+ Files.createFile(Paths.get(markerPath));
+ return true;
+ } catch (FileAlreadyExistsException ignored) {
+ return false;
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ static void initializeSchema(String evidencePath, String schema) {
+ Path schemaPath = Paths.get(evidencePath + ".schema");
+ if (!Files.exists(schemaPath)) {
+ writeSchema(evidencePath, schema);
+ }
+ }
+
+ static void writeSchema(String evidencePath, String schema) {
+ try {
+ Files.write(
+ Paths.get(evidencePath + ".schema"),
+ schema.getBytes(StandardCharsets.UTF_8),
+ StandardOpenOption.CREATE,
+ StandardOpenOption.TRUNCATE_EXISTING);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+}
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
index 65e8448f1f..2ee0b97c75 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
@@ -154,6 +154,10 @@ public abstract class PipelineTestEnvironment extends
TestLogger {
"taskmanager.memory.jvm-metaspace.size: 512mb");
public static final String FLINK_PROPERTIES = String.join("\n",
EXTERNAL_PROPS);
+ protected String getFlinkProperties() {
+ return FLINK_PROPERTIES;
+ }
+
@Nullable protected RestClusterClient<StandaloneClusterId>
restClusterClient;
protected GenericContainer<?> jobManager;
@@ -189,7 +193,7 @@ public abstract class PipelineTestEnvironment extends
TestLogger {
.withNetwork(NETWORK)
.withNetworkAliases(INTER_CONTAINER_JM_ALIAS)
.withExposedPorts(JOB_MANAGER_REST_PORT)
- .withEnv("FLINK_PROPERTIES", FLINK_PROPERTIES)
+ .withEnv("FLINK_PROPERTIES", getFlinkProperties())
.withCreateContainerCmdModifier(cmd ->
cmd.withVolumes(sharedVolume))
.withLogConsumer(jobManagerConsumer);
@@ -211,7 +215,7 @@ public abstract class PipelineTestEnvironment extends
TestLogger {
.withCommand("taskmanager")
.withNetwork(NETWORK)
.withNetworkAliases(INTER_CONTAINER_TM_ALIAS)
- .withEnv("FLINK_PROPERTIES", FLINK_PROPERTIES)
+ .withEnv("FLINK_PROPERTIES", getFlinkProperties())
.dependsOn(jobManager)
.withVolumesFrom(jobManager, BindMode.READ_WRITE)
.withLogConsumer(taskManagerConsumer);
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java
index af0cd6f4cb..5dc1ff99e6 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java
@@ -55,6 +55,8 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BooleanSupplier;
@@ -100,6 +102,10 @@ public abstract class SchemaRegistry implements
OperatorCoordinator, Coordinatio
protected transient SchemaManager schemaManager;
protected transient TableIdRouter router;
+ private final Object lifecycleLock = new Object();
+ private volatile boolean resetting;
+ private volatile long generation;
+
protected SchemaRegistry(
OperatorCoordinator.Context context,
String operatorName,
@@ -125,6 +131,13 @@ public abstract class SchemaRegistry implements
OperatorCoordinator, Coordinatio
@Override
public void start() throws Exception {
LOG.info("Starting SchemaRegistry - {}.", operatorName);
+ initializeBaseRuntimeState();
+ initialize();
+ LOG.info(
+ "Started SchemaRegistry for {}. Parallelism: {}",
operatorName, currentParallelism);
+ }
+
+ private void initializeBaseRuntimeState() {
this.currentParallelism = context.currentParallelism();
this.activeSinkWriters = ConcurrentHashMap.newKeySet();
this.failedReasons = new ConcurrentHashMap<>();
@@ -155,6 +168,15 @@ public abstract class SchemaRegistry implements
OperatorCoordinator, Coordinatio
/** Restore schema registry state from byte array. */
protected abstract void restore(byte[] checkpointData) throws Exception;
+ /**
+ * Stops schema-change work from the previous coordinator generation.
Implementations must block
+ * until the previous generation's worker has fully exited, within {@link
#rpcTimeout}.
+ */
+ protected abstract void shutdown() throws Exception;
+
+ /** (Re)initializes transient state, both on {@link #start()} and after a
coordinator reset. */
+ protected abstract void initialize();
+
// ------------------------------------
// Overridable event & request handlers
// ------------------------------------
@@ -258,6 +280,7 @@ public abstract class SchemaRegistry implements
OperatorCoordinator, Coordinatio
handleCustomCoordinationRequest(request, future);
}
},
+ future,
"Handling request - %s",
request);
return future;
@@ -313,7 +336,11 @@ public abstract class SchemaRegistry implements
OperatorCoordinator, Coordinatio
public final void checkpointCoordinator(
long checkpointId, CompletableFuture<byte[]> completableFuture)
throws Exception {
LOG.info("Going to start checkpoint No.{}", checkpointId);
- runInEventLoop(() -> snapshot(completableFuture), "Taking checkpoint -
%d", checkpointId);
+ runInEventLoop(
+ () -> snapshot(completableFuture),
+ completableFuture,
+ "Taking checkpoint - %d",
+ checkpointId);
}
@Override
@@ -325,10 +352,22 @@ public abstract class SchemaRegistry implements
OperatorCoordinator, Coordinatio
public final void resetToCheckpoint(long checkpointId, @Nullable byte[]
checkpointData)
throws Exception {
LOG.info("Going to restore from checkpoint No.{}", checkpointId);
+ synchronized (lifecycleLock) {
+ resetting = true;
+ generation++;
+ }
+ awaitCoordinatorExecutor();
+ shutdown();
if (checkpointData == null) {
- return;
+ schemaManager = new SchemaManager();
+ } else {
+ restore(checkpointData);
+ }
+ initializeBaseRuntimeState();
+ initialize();
+ synchronized (lifecycleLock) {
+ resetting = false;
}
- restore(checkpointData);
}
// ---------------------------
@@ -344,21 +383,65 @@ public abstract class SchemaRegistry implements
OperatorCoordinator, Coordinatio
final ThrowingRunnable<Throwable> action,
final String actionName,
final Object... actionNameFormatParameters) {
+ runInEventLoop(action, null, actionName, actionNameFormatParameters);
+ }
+
+ private void runInEventLoop(
+ final ThrowingRunnable<Throwable> action,
+ @Nullable CompletableFuture<?> cancellationFuture,
+ final String actionName,
+ final Object... actionNameFormatParameters) {
+ final long actionGeneration;
+ synchronized (lifecycleLock) {
+ if (resetting) {
+ reject(cancellationFuture);
+ return;
+ }
+ actionGeneration = generation;
+ }
coordinatorExecutor.execute(
() -> {
+ synchronized (lifecycleLock) {
+ if (resetting || actionGeneration != generation) {
+ reject(cancellationFuture);
+ return;
+ }
+ }
try {
action.run();
} catch (Throwable t) {
- // if we have a JVM critical error, promote it
immediately, there is a good
+ // if we have a JVM critical error, promote it
immediately, there is a
+ // good
// chance the logging or job failing will not succeed
anymore
ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
- handleUnrecoverableError(
- String.format(actionName,
actionNameFormatParameters), t);
- context.failJob(t);
+ synchronized (lifecycleLock) {
+ if (resetting || actionGeneration != generation) {
+ reject(cancellationFuture);
+ return;
+ }
+ failJob(String.format(actionName,
actionNameFormatParameters), t);
+ }
}
});
}
+ private void awaitCoordinatorExecutor() throws Exception {
+ Future<?> barrier = coordinatorExecutor.submit(() -> {});
+ try {
+ barrier.get(rpcTimeout.toMillis(), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw e;
+ }
+ }
+
+ private void reject(@Nullable CompletableFuture<?> future) {
+ if (future != null) {
+ future.completeExceptionally(
+ new FlinkRuntimeException("Schema coordinator is
resetting."));
+ }
+ }
+
/**
* Keeps checking if {@code conditionChecker} is satisfied. If not, emit a
message and retry.
*/
@@ -391,10 +474,15 @@ public abstract class SchemaRegistry implements
OperatorCoordinator, Coordinatio
}
protected <T extends Throwable> void failJob(String taskDescription, T t) {
- ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
- LOG.error("An exception was triggered from {}. Job will fail now.",
taskDescription, t);
- handleUnrecoverableError(taskDescription, t);
- context.failJob(t);
+ synchronized (lifecycleLock) {
+ ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
+ if (resetting) {
+ return;
+ }
+ LOG.error("An exception was triggered from {}. Job will fail
now.", taskDescription, t);
+ handleUnrecoverableError(taskDescription, t);
+ context.failJob(t);
+ }
}
// ------------------------
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java
index f9aad19b85..286f89bfe7 100755
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java
@@ -64,6 +64,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
@@ -102,7 +103,7 @@ public class SchemaCoordinator extends SchemaRegistry {
alreadyHandledSchemaChangeEvents;
/** Executor service to execute schema change. */
- private final ExecutorService schemaChangeThreadPool;
+ private transient ExecutorService schemaChangeThreadPool;
public SchemaCoordinator(
String operatorName,
@@ -129,15 +130,40 @@ public class SchemaCoordinator extends SchemaRegistry {
// Lifecycle methods
// -----------------
@Override
- public void start() throws Exception {
- super.start();
+ protected void initialize() {
+ if (pendingRequests != null) {
+ pendingRequests.forEach(
+ (index, tuple) ->
+ tuple.f1.completeExceptionally(
+ new FlinkRuntimeException(
+ "Schema coordinator request was
cancelled by checkpoint reset.")));
+ }
this.evolvingStatus = new AtomicReference<>(RequestStatus.IDLE);
this.pendingRequests = new ConcurrentHashMap<>();
this.flushedSinkWriters = ConcurrentHashMap.newKeySet();
this.upstreamSchemaTable = HashBasedTable.create();
this.alreadyHandledSchemaChangeEvents = HashMultimap.create();
- LOG.info(
- "Started SchemaRegistry for {}. Parallelism: {}",
operatorName, currentParallelism);
+ if (schemaChangeThreadPool == null ||
schemaChangeThreadPool.isShutdown()) {
+ schemaChangeThreadPool = Executors.newSingleThreadExecutor();
+ }
+ }
+
+ @Override
+ protected void shutdown() throws Exception {
+ schemaChangeThreadPool.shutdownNow();
+ if (!schemaChangeThreadPool.awaitTermination(
+ rpcTimeout.toMillis(), TimeUnit.MILLISECONDS)) {
+ throw new TimeoutException("Schema change executor did not
terminate during reset.");
+ }
+ }
+
+ @Override
+ protected void handleUnrecoverableError(String taskDescription, Throwable
t) {
+ super.handleUnrecoverableError(taskDescription, t);
+ if (pendingRequests != null) {
+ pendingRequests.forEach((index, tuple) ->
tuple.f1.completeExceptionally(t));
+ }
+ LOG.info("Current upstream table state: {}", upstreamSchemaTable);
}
@Override
@@ -208,16 +234,6 @@ public class SchemaCoordinator extends SchemaRegistry {
flushedSinkWriters.add(event.getSinkSubTaskId());
}
- @Override
- protected void handleUnrecoverableError(String taskDescription, Throwable
t) {
- super.handleUnrecoverableError(taskDescription, t);
- LOG.info("Current upstream table state: {}", upstreamSchemaTable);
- pendingRequests.forEach(
- (index, tuple) -> {
- tuple.f1.completeExceptionally(t);
- });
- }
-
// -------------------------
// Schema evolving logic
// -------------------------
@@ -472,12 +488,8 @@ public class SchemaCoordinator extends SchemaRegistry {
schemaChangeEvent);
return true;
} catch (Throwable t) {
- handleUnrecoverableError(
- "Apply schema change event - " + schemaChangeEvent,
- new FlinkRuntimeException(
- "Failed to apply schema change event " +
schemaChangeEvent + ".", t));
- context.failJob(t);
- throw t;
+ throw new FlinkRuntimeException(
+ "Failed to apply schema change event " + schemaChangeEvent
+ ".", t);
}
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java
index 3a241aac48..da7d78e306 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java
@@ -64,6 +64,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
@@ -74,7 +75,7 @@ public class SchemaCoordinator extends SchemaRegistry {
private static final Logger LOG =
LoggerFactory.getLogger(SchemaCoordinator.class);
/** Executor service to execute schema change. */
- private final ExecutorService schemaChangeThreadPool;
+ private transient ExecutorService schemaChangeThreadPool;
/**
* Sink writers which have sent flush success events for the request.<br>
@@ -117,10 +118,28 @@ public class SchemaCoordinator extends SchemaRegistry {
}
@Override
- public void start() throws Exception {
- super.start();
+ protected void initialize() {
+ if (pendingRequests != null) {
+ pendingRequests.forEach(
+ (index, tuple) ->
+ tuple.f1.completeExceptionally(
+ new FlinkRuntimeException(
+ "Schema coordinator request was
cancelled by checkpoint reset.")));
+ }
this.flushedSinkWriters = new ConcurrentHashMap<>();
this.pendingRequests = new ConcurrentHashMap<>();
+ if (schemaChangeThreadPool == null ||
schemaChangeThreadPool.isShutdown()) {
+ schemaChangeThreadPool = Executors.newSingleThreadExecutor();
+ }
+ }
+
+ @Override
+ protected void shutdown() throws Exception {
+ schemaChangeThreadPool.shutdownNow();
+ if (!schemaChangeThreadPool.awaitTermination(
+ rpcTimeout.toMillis(), TimeUnit.MILLISECONDS)) {
+ throw new TimeoutException("Schema change executor did not
terminate during reset.");
+ }
}
@Override
@@ -227,10 +246,9 @@ public class SchemaCoordinator extends SchemaRegistry {
super.handleUnrecoverableError(taskDescription, t);
// For each pending future, release it exceptionally before quitting
- pendingRequests.forEach(
- (index, tuple) -> {
- tuple.f1.completeExceptionally(t);
- });
+ if (pendingRequests != null) {
+ pendingRequests.forEach((index, tuple) ->
tuple.f1.completeExceptionally(t));
+ }
}
/**
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinatorTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinatorTest.java
new file mode 100644
index 0000000000..35d92e1764
--- /dev/null
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinatorTest.java
@@ -0,0 +1,426 @@
+/*
+ * 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.cdc.runtime.operators.schema.distributed;
+
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.exceptions.SchemaEvolveException;
+import org.apache.flink.cdc.common.pipeline.RouteMode;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.utils.SchemaUtils;
+import
org.apache.flink.cdc.runtime.operators.schema.common.event.FlushSuccessEvent;
+import
org.apache.flink.cdc.runtime.operators.schema.common.event.GetEvolvedSchemaRequest;
+import
org.apache.flink.cdc.runtime.operators.schema.common.event.GetEvolvedSchemaResponse;
+import
org.apache.flink.cdc.runtime.operators.schema.distributed.event.SchemaChangeRequest;
+import
org.apache.flink.cdc.runtime.operators.schema.distributed.event.SchemaChangeResponse;
+import
org.apache.flink.cdc.runtime.testutils.operators.MockedOperatorCoordinatorContext;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.operators.coordination.CoordinationResponse;
+import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static
org.apache.flink.cdc.runtime.operators.schema.common.CoordinationResponseUtils.unwrap;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link SchemaCoordinator}. */
+class SchemaCoordinatorTest {
+
+ private static final TableId TABLE_ID =
TableId.parse("db.schema_recovery");
+ private static final Schema INITIAL_SCHEMA =
+ Schema.newBuilder().physicalColumn("id",
DataTypes.INT().notNull()).build();
+ private static final AddColumnEvent ADD_EXTRA_V2 =
+ new AddColumnEvent(
+ TABLE_ID,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("extra_v2",
DataTypes.STRING()),
+ AddColumnEvent.ColumnPosition.LAST,
+ null)));
+ private static final Schema EVOLVED_SCHEMA =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("extra_v2", DataTypes.STRING())
+ .build();
+
+ @Test
+ void resetsFailedSchemaEvolutionBeforeReplayingSchemaEvents() throws
Exception {
+ FailOnceMetadataApplier metadataApplier = new
FailOnceMetadataApplier();
+ ExecutorService coordinatorExecutor =
Executors.newSingleThreadExecutor();
+ SchemaCoordinator coordinator =
+ new SchemaCoordinator(
+ "schema-coordinator",
+ new MockedOperatorCoordinatorContext(
+ new OperatorID(),
Thread.currentThread().getContextClassLoader()),
+ coordinatorExecutor,
+ metadataApplier,
+ Collections.emptyList(),
+ RouteMode.ALL_MATCH,
+ SchemaChangeBehavior.LENIENT,
+ Duration.ofSeconds(10));
+ coordinator.start();
+
+ try {
+ requestSchemaChange(coordinator, new CreateTableEvent(TABLE_ID,
INITIAL_SCHEMA));
+ byte[] checkpoint = checkpoint(coordinator);
+
+ CompletableFuture<CoordinationResponse> failedRequest =
+ submitSchemaChange(coordinator, ADD_EXTRA_V2);
+ assertThat(metadataApplier.failedAdd.await(10,
TimeUnit.SECONDS)).isTrue();
+ assertThatThrownBy(() -> failedRequest.get(10, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class);
+
+ coordinator.resetToCheckpoint(1L, checkpoint);
+
+ requestSchemaChange(coordinator, new CreateTableEvent(TABLE_ID,
INITIAL_SCHEMA));
+ SchemaChangeResponse response = requestSchemaChange(coordinator,
ADD_EXTRA_V2);
+
+ assertThat(response.getEvolvedSchemas()).containsEntry(TABLE_ID,
EVOLVED_SCHEMA);
+
assertThat(response.getEvolvedSchemaChangeEvents()).containsExactly(ADD_EXTRA_V2);
+
assertThat(metadataApplier.physicalSchemas).containsEntry(TABLE_ID,
EVOLVED_SCHEMA);
+ } finally {
+ coordinator.close();
+ }
+ }
+
+ @Test
+ void nullCheckpointClearsSchemaStateBeforeReplay() throws Exception {
+ FailOnceMetadataApplier metadataApplier = new
FailOnceMetadataApplier(false);
+ SchemaCoordinator coordinator = createCoordinator(metadataApplier,
Duration.ofSeconds(10));
+ coordinator.start();
+
+ try {
+ requestSchemaChange(coordinator, new CreateTableEvent(TABLE_ID,
INITIAL_SCHEMA));
+ coordinator.resetToCheckpoint(1L, null);
+
+ SchemaChangeResponse response =
+ requestSchemaChange(
+ coordinator, new CreateTableEvent(TABLE_ID,
INITIAL_SCHEMA));
+
+ assertThat(response.getEvolvedSchemaChangeEvents())
+ .containsExactly(new CreateTableEvent(TABLE_ID,
INITIAL_SCHEMA));
+ assertThat(metadataApplier.appliedEvents)
+ .containsExactly(
+ new CreateTableEvent(TABLE_ID, INITIAL_SCHEMA),
+ new CreateTableEvent(TABLE_ID, INITIAL_SCHEMA));
+ } finally {
+ coordinator.close();
+ }
+ }
+
+ @Test
+ void restoresCheckpointBeforeStart() throws Exception {
+ FailOnceMetadataApplier sourceApplier = new
FailOnceMetadataApplier(false);
+ SchemaCoordinator source = createCoordinator(sourceApplier,
Duration.ofSeconds(10));
+ source.start();
+ byte[] checkpoint;
+ try {
+ requestSchemaChange(source, new CreateTableEvent(TABLE_ID,
INITIAL_SCHEMA));
+ checkpoint = checkpoint(source);
+ } finally {
+ source.close();
+ }
+
+ SchemaCoordinator restored =
+ createCoordinator(new FailOnceMetadataApplier(false),
Duration.ofSeconds(10));
+ restored.resetToCheckpoint(1L, checkpoint);
+ restored.start();
+ try {
+ GetEvolvedSchemaResponse response =
+ unwrap(
+ restored.handleCoordinationRequest(
+ new GetEvolvedSchemaRequest(
+ TABLE_ID,
+
GetEvolvedSchemaRequest.LATEST_SCHEMA_VERSION))
+ .get(10, TimeUnit.SECONDS));
+
+ assertThat(response.getSchema()).contains(INITIAL_SCHEMA);
+ } finally {
+ restored.close();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void resetWaitsForAnUninterruptibleSchemaChangeBeforeRetrying(boolean
failLate)
+ throws Exception {
+ FailOnceMetadataApplier metadataApplier =
+ new FailOnceMetadataApplier(false, true, failLate);
+ CountingCoordinatorContext context = new CountingCoordinatorContext();
+ SchemaCoordinator coordinator =
+ createCoordinator(metadataApplier, context,
Duration.ofMillis(200));
+ coordinator.start();
+
+ try {
+ requestSchemaChange(coordinator, new CreateTableEvent(TABLE_ID,
INITIAL_SCHEMA));
+ CompletableFuture<CoordinationResponse> oldRequest =
+ submitSchemaChange(coordinator, ADD_EXTRA_V2);
+ assertThat(metadataApplier.blockedAdd.await(10,
TimeUnit.SECONDS)).isTrue();
+
+ assertThatThrownBy(() -> coordinator.resetToCheckpoint(1L, null))
+ .isInstanceOf(TimeoutException.class);
+ CompletableFuture<CoordinationResponse> rejectedRequest =
+ coordinator.handleCoordinationRequest(
+ new GetEvolvedSchemaRequest(
+ TABLE_ID,
GetEvolvedSchemaRequest.LATEST_SCHEMA_VERSION));
+ CompletableFuture<byte[]> rejectedCheckpoint = new
CompletableFuture<>();
+ coordinator.checkpointCoordinator(2L, rejectedCheckpoint);
+ assertThatThrownBy(() -> rejectedRequest.get(1, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class);
+ assertThatThrownBy(() -> rejectedCheckpoint.get(1,
TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class);
+ assertThat(metadataApplier.appliedEvents).hasSize(1);
+
+ metadataApplier.releaseAdd.countDown();
+ coordinator.resetToCheckpoint(1L, null);
+ if (failLate) {
+ assertThatThrownBy(() -> oldRequest.get(10, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class);
+ } else {
+ oldRequest.get(10, TimeUnit.SECONDS);
+ }
+ assertThat(context.failures.get()).isZero();
+
+ requestSchemaChange(coordinator, new CreateTableEvent(TABLE_ID,
INITIAL_SCHEMA));
+ requestSchemaChange(coordinator, ADD_EXTRA_V2);
+ } finally {
+ metadataApplier.releaseAdd.countDown();
+ coordinator.close();
+ }
+ }
+
+ @Test
+ void rejectsQueuedRequestAndCheckpointAcrossReset() throws Exception {
+ CountDownLatch releaseExecutor = new CountDownLatch(1);
+ CountingExecutor coordinatorExecutor = new
CountingExecutor(releaseExecutor);
+ FailOnceMetadataApplier metadataApplier = new
FailOnceMetadataApplier(false);
+ SchemaCoordinator coordinator =
+ new SchemaCoordinator(
+ "schema-coordinator",
+ new CountingCoordinatorContext(),
+ coordinatorExecutor,
+ metadataApplier,
+ Collections.emptyList(),
+ RouteMode.ALL_MATCH,
+ SchemaChangeBehavior.LENIENT,
+ Duration.ofSeconds(2));
+ AtomicReference<Throwable> resetFailure = new AtomicReference<>();
+ SchemaCoordinator resettingCoordinator = coordinator;
+ Thread resetThread =
+ new Thread(
+ () -> {
+ try {
+ resettingCoordinator.resetToCheckpoint(1L,
null);
+ } catch (Throwable t) {
+ resetFailure.set(t);
+ }
+ });
+ try {
+ coordinator.start();
+ CompletableFuture<CoordinationResponse> request =
+ coordinator.handleCoordinationRequest(
+ new SchemaChangeRequest(
+ 0, 0, new CreateTableEvent(TABLE_ID,
INITIAL_SCHEMA)));
+ CompletableFuture<byte[]> checkpoint = new CompletableFuture<>();
+ coordinator.checkpointCoordinator(1L, checkpoint);
+ resetThread.start();
+ assertThat(coordinatorExecutor.barrierSubmitted.await(10,
TimeUnit.SECONDS)).isTrue();
+ releaseExecutor.countDown();
+ resetThread.join(10_000);
+ assertThat(resetThread.isAlive()).isFalse();
+ assertThat(resetFailure.get()).isNull();
+ assertThatThrownBy(() -> request.get(1, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class);
+ assertThatThrownBy(() -> checkpoint.get(1, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class);
+ assertThat(metadataApplier.appliedEvents).isEmpty();
+ } finally {
+ releaseExecutor.countDown();
+ resetThread.join(10_000);
+ coordinator.close();
+ }
+ }
+
+ private static SchemaCoordinator createCoordinator(
+ FailOnceMetadataApplier metadataApplier, Duration rpcTimeout) {
+ return createCoordinator(
+ metadataApplier,
+ new MockedOperatorCoordinatorContext(
+ new OperatorID(),
Thread.currentThread().getContextClassLoader()),
+ rpcTimeout);
+ }
+
+ private static SchemaCoordinator createCoordinator(
+ FailOnceMetadataApplier metadataApplier,
+ OperatorCoordinator.Context context,
+ Duration rpcTimeout) {
+ return new SchemaCoordinator(
+ "schema-coordinator",
+ context,
+ Executors.newSingleThreadExecutor(),
+ metadataApplier,
+ Collections.emptyList(),
+ RouteMode.ALL_MATCH,
+ SchemaChangeBehavior.LENIENT,
+ rpcTimeout);
+ }
+
+ private static byte[] checkpoint(SchemaCoordinator coordinator) throws
Exception {
+ CompletableFuture<byte[]> checkpoint = new CompletableFuture<>();
+ coordinator.checkpointCoordinator(1L, checkpoint);
+ return checkpoint.get(10, TimeUnit.SECONDS);
+ }
+
+ private static SchemaChangeResponse requestSchemaChange(
+ SchemaCoordinator coordinator, SchemaChangeEvent event) throws
Exception {
+ return unwrap(submitSchemaChange(coordinator, event).get(10,
TimeUnit.SECONDS));
+ }
+
+ private static CompletableFuture<CoordinationResponse> submitSchemaChange(
+ SchemaCoordinator coordinator, SchemaChangeEvent event) {
+ CompletableFuture<CoordinationResponse> response =
+ coordinator.handleCoordinationRequest(new
SchemaChangeRequest(0, 0, event));
+ coordinator.handleEventFromOperator(0, 0, new FlushSuccessEvent(0, 0));
+ return response;
+ }
+
+ private static final class FailOnceMetadataApplier implements
MetadataApplier {
+ private final AtomicBoolean failNextAdd;
+ private final CountDownLatch failedAdd = new CountDownLatch(1);
+ private final CountDownLatch blockedAdd = new CountDownLatch(1);
+ private final CountDownLatch releaseAdd = new CountDownLatch(1);
+ private final Map<TableId, Schema> physicalSchemas = new
ConcurrentHashMap<>();
+ private final List<SchemaChangeEvent> appliedEvents = new
ArrayList<>();
+ private final boolean blockAdd;
+ private final AtomicBoolean failLate;
+
+ private FailOnceMetadataApplier() {
+ this(true);
+ }
+
+ private FailOnceMetadataApplier(boolean failNextAdd) {
+ this(failNextAdd, false);
+ }
+
+ private FailOnceMetadataApplier(boolean failNextAdd, boolean blockAdd)
{
+ this(failNextAdd, blockAdd, false);
+ }
+
+ private FailOnceMetadataApplier(boolean failNextAdd, boolean blockAdd,
boolean failLate) {
+ this.failNextAdd = new AtomicBoolean(failNextAdd);
+ this.blockAdd = blockAdd;
+ this.failLate = new AtomicBoolean(failLate);
+ }
+
+ @Override
+ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) {
+ if (schemaChangeEvent instanceof AddColumnEvent
+ && failNextAdd.compareAndSet(true, false)) {
+ failedAdd.countDown();
+ throw new SchemaEvolveException(schemaChangeEvent, "Injected
metadata failure");
+ }
+ if (schemaChangeEvent instanceof AddColumnEvent && blockAdd) {
+ blockedAdd.countDown();
+ while (releaseAdd.getCount() > 0) {
+ try {
+ releaseAdd.await();
+ } catch (InterruptedException ignored) {
+ // Simulates an external DDL call that ignores
interruption.
+ }
+ }
+ }
+ if (schemaChangeEvent instanceof AddColumnEvent
+ && failLate.compareAndSet(true, false)) {
+ throw new SchemaEvolveException(
+ schemaChangeEvent, "Injected late metadata failure");
+ }
+ physicalSchemas.put(
+ schemaChangeEvent.tableId(),
+ SchemaUtils.applySchemaChangeEvent(
+ physicalSchemas.get(schemaChangeEvent.tableId()),
schemaChangeEvent));
+ appliedEvents.add(schemaChangeEvent);
+ }
+ }
+
+ private static final class CountingCoordinatorContext extends
MockedOperatorCoordinatorContext {
+ private final AtomicInteger failures = new AtomicInteger();
+
+ private CountingCoordinatorContext() {
+ super(new OperatorID(),
Thread.currentThread().getContextClassLoader());
+ }
+
+ @Override
+ public void failJob(Throwable cause) {
+ failures.incrementAndGet();
+ super.failJob(cause);
+ }
+ }
+
+ private static final class CountingExecutor extends ThreadPoolExecutor {
+ private final AtomicInteger submissions = new AtomicInteger();
+ private final CountDownLatch barrierSubmitted = new CountDownLatch(1);
+
+ private CountingExecutor(CountDownLatch release) {
+ super(1, 1, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
+ execute(
+ () -> {
+ try {
+ release.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ if (submissions.incrementAndGet() == 4) {
+ barrierSubmitted.countDown();
+ }
+ super.execute(command);
+ }
+ }
+}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinatorTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinatorTest.java
new file mode 100644
index 0000000000..a30127f510
--- /dev/null
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinatorTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.cdc.runtime.operators.schema.regular;
+
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.exceptions.SchemaEvolveException;
+import org.apache.flink.cdc.common.pipeline.RouteMode;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.cdc.common.types.DataTypes;
+import
org.apache.flink.cdc.runtime.operators.schema.common.event.FlushSuccessEvent;
+import
org.apache.flink.cdc.runtime.operators.schema.regular.event.SchemaChangeRequest;
+import
org.apache.flink.cdc.runtime.operators.schema.regular.event.SchemaChangeResponse;
+import
org.apache.flink.cdc.runtime.testutils.operators.MockedOperatorCoordinatorContext;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.operators.coordination.CoordinationResponse;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static
org.apache.flink.cdc.runtime.operators.schema.common.CoordinationResponseUtils.unwrap;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the regular {@link SchemaCoordinator}. */
+class SchemaCoordinatorTest {
+
+ private static final TableId TABLE_ID =
TableId.parse("db.regular_recovery");
+ private static final Schema INITIAL_SCHEMA =
+ Schema.newBuilder().physicalColumn("id",
DataTypes.INT().notNull()).build();
+ private static final AddColumnEvent ADD_EXTRA_V2 =
+ new AddColumnEvent(
+ TABLE_ID,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("extra_v2",
DataTypes.STRING()),
+ AddColumnEvent.ColumnPosition.LAST,
+ null)));
+
+ @Test
+ void resetsFailedSchemaEvolutionBeforeReplay() throws Exception {
+ FailOnceMetadataApplier metadataApplier = new
FailOnceMetadataApplier();
+ SchemaCoordinator coordinator =
+ new SchemaCoordinator(
+ "regular-schema-coordinator",
+ new MockedOperatorCoordinatorContext(
+ new OperatorID(),
Thread.currentThread().getContextClassLoader()),
+ Executors.newSingleThreadExecutor(),
+ metadataApplier,
+ Collections.emptyList(),
+ RouteMode.ALL_MATCH,
+ SchemaChangeBehavior.LENIENT,
+ Duration.ofSeconds(10));
+ coordinator.start();
+
+ try {
+ requestSchemaChange(coordinator, new CreateTableEvent(TABLE_ID,
INITIAL_SCHEMA));
+ byte[] checkpoint = checkpoint(coordinator);
+
+ CompletableFuture<CoordinationResponse> failedRequest =
+ submitSchemaChange(coordinator, ADD_EXTRA_V2);
+ assertThat(metadataApplier.failedAdd.await(10,
TimeUnit.SECONDS)).isTrue();
+ assertThatThrownBy(() -> failedRequest.get(10, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class);
+
+ coordinator.resetToCheckpoint(1L, checkpoint);
+
+ requestSchemaChange(coordinator, new CreateTableEvent(TABLE_ID,
INITIAL_SCHEMA));
+ SchemaChangeResponse response = requestSchemaChange(coordinator,
ADD_EXTRA_V2);
+
+
assertThat(response.getAppliedSchemaChangeEvents()).containsExactly(ADD_EXTRA_V2);
+ } finally {
+ coordinator.close();
+ }
+ }
+
+ private static byte[] checkpoint(SchemaCoordinator coordinator) throws
Exception {
+ CompletableFuture<byte[]> checkpoint = new CompletableFuture<>();
+ coordinator.checkpointCoordinator(1L, checkpoint);
+ return checkpoint.get(10, TimeUnit.SECONDS);
+ }
+
+ private static SchemaChangeResponse requestSchemaChange(
+ SchemaCoordinator coordinator, SchemaChangeEvent event) throws
Exception {
+ return unwrap(submitSchemaChange(coordinator, event).get(10,
TimeUnit.SECONDS));
+ }
+
+ private static CompletableFuture<CoordinationResponse> submitSchemaChange(
+ SchemaCoordinator coordinator, SchemaChangeEvent event) {
+ CompletableFuture<CoordinationResponse> response =
+ coordinator.handleCoordinationRequest(new
SchemaChangeRequest(TABLE_ID, event, 0));
+ coordinator.handleEventFromOperator(0, 0, new FlushSuccessEvent(0, 0));
+ return response;
+ }
+
+ private static final class FailOnceMetadataApplier implements
MetadataApplier {
+ private final AtomicBoolean failNextAdd = new AtomicBoolean(true);
+ private final CountDownLatch failedAdd = new CountDownLatch(1);
+
+ @Override
+ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) {
+ if (schemaChangeEvent instanceof AddColumnEvent
+ && failNextAdd.compareAndSet(true, false)) {
+ failedAdd.countDown();
+ throw new SchemaEvolveException(schemaChangeEvent, "Injected
metadata failure");
+ }
+ }
+ }
+}
