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 391169d05 [hotfix][runtime Preserve original failure during
distributed schema evolution (#4556)
391169d05 is described below
commit 391169d059b54f538b1e698e51ba3c36da69d5cc
Author: yuxiqian <[email protected]>
AuthorDate: Wed Sep 23 17:56:12 2026 +0800
[hotfix][runtime Preserve original failure during distributed schema
evolution (#4556)
Co-authored-by: lvyanquan.lyq <[email protected]>
---
...edSchemaEvolutionFailurePreservationITCase.java | 162 ++++++++++++++
.../testsink/factory/FailingDataSinkFactory.java | 224 +++++++++++++++++++
.../org.apache.flink.cdc.common.factories.Factory | 1 +
.../operators/schema/common/SchemaRegistry.java | 2 +-
.../schema/distributed/SchemaCoordinator.java | 71 +++++-
.../schema/distributed/SchemaCoordinatorTest.java | 242 +++++++++++++++++++++
6 files changed, 697 insertions(+), 5 deletions(-)
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/DistributedSchemaEvolutionFailurePreservationITCase.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/DistributedSchemaEvolutionFailurePreservationITCase.java
new file mode 100644
index 000000000..2f86709d7
--- /dev/null
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/DistributedSchemaEvolutionFailurePreservationITCase.java
@@ -0,0 +1,162 @@
+/*
+ * 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.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
+import org.apache.flink.cdc.composer.PipelineExecution;
+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.testsink.factory.FailingDataSinkFactory;
+import
org.apache.flink.cdc.composer.testsource.factory.DistributedDataSourceFactory;
+import
org.apache.flink.cdc.composer.testsource.source.DistributedSourceOptions;
+import org.apache.flink.configuration.RestartStrategyOptions;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static
org.apache.flink.configuration.CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL;
+
+/** Integration tests for preserving failures during distributed schema
evolution. */
+@Timeout(value = 120, unit = TimeUnit.SECONDS)
+class DistributedSchemaEvolutionFailurePreservationITCase {
+
+ private static final Duration EVENT_TIMEOUT = Duration.ofSeconds(30);
+
+ 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"));
+ MINI_CLUSTER_CONFIG.set(RestartStrategyOptions.RESTART_STRATEGY,
"fixed-delay");
+
MINI_CLUSTER_CONFIG.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS,
1);
+ MINI_CLUSTER_CONFIG.set(
+ RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY,
Duration.ZERO);
+ }
+
+ @RegisterExtension
+ static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(2)
+ .setConfiguration(MINI_CLUSTER_CONFIG)
+ .build());
+
+ @BeforeEach
+ void setup() {
+ FailingDataSinkFactory.reset();
+ }
+
+ @AfterEach
+ void cleanup() {
+ FailingDataSinkFactory.releaseMetadataApplier();
+ }
+
+ @Test
+ void testPreserveSinkFailureDuringDistributedSchemaEvolution() throws
Exception {
+ PipelineExecution execution = createPipelineExecution();
+ CompletableFuture<Throwable> executionFailure =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ execution.execute();
+ throw new AssertionError("Pipeline execution
should fail.");
+ } catch (Throwable t) {
+ return t;
+ }
+ });
+
+ try {
+
Assertions.assertThat(FailingDataSinkFactory.awaitMetadataApplying(EVENT_TIMEOUT))
+ .as("Metadata applier should block while the coordinator
is EVOLVING.")
+ .isTrue();
+
Assertions.assertThat(FailingDataSinkFactory.awaitSinkFailure(EVENT_TIMEOUT))
+ .as("A sink subtask should fail while schema evolution is
in progress.")
+ .isTrue();
+
+ Throwable failure =
+ executionFailure.get(EVENT_TIMEOUT.toMillis(),
TimeUnit.MILLISECONDS);
+ Assertions.assertThat(failure)
+ .as("The original sink failure should remain the root
cause.")
+
.hasRootCauseMessage(FailingDataSinkFactory.ORIGINAL_FAILURE_MESSAGE);
+ Assertions.assertThat(getRootCause(failure).getSuppressed())
+ .as(
+ "The schema evolution state error should only be
attached as a suppressed exception.")
+ .anySatisfy(
+ suppressed ->
+ Assertions.assertThat(suppressed)
+
.isInstanceOf(IllegalStateException.class)
+ .hasMessage("Unexpected evolving
status: EVOLVING"));
+ } finally {
+ FailingDataSinkFactory.releaseMetadataApplier();
+ executionFailure.cancel(true);
+ }
+ }
+
+ private static PipelineExecution createPipelineExecution() {
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(DistributedSourceOptions.TABLE_COUNT, 1);
+ sourceConfig.set(DistributedSourceOptions.DISTRIBUTED_TABLES, true);
+ SourceDef sourceDef =
+ new SourceDef(
+ DistributedDataSourceFactory.IDENTIFIER,
+ "Distributed Source",
+ sourceConfig);
+
+ SinkDef sinkDef =
+ new SinkDef(FailingDataSinkFactory.IDENTIFIER, "Failing Sink",
new Configuration());
+
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR,
SchemaChangeBehavior.LENIENT);
+ return FlinkPipelineComposer.ofMiniCluster()
+ .compose(
+ new PipelineDef(
+ sourceDef,
+ sinkDef,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ pipelineConfig));
+ }
+
+ private static Throwable getRootCause(Throwable throwable) {
+ Throwable rootCause = throwable;
+ while (rootCause.getCause() != null) {
+ rootCause = rootCause.getCause();
+ }
+ return rootCause;
+ }
+}
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsink/factory/FailingDataSinkFactory.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsink/factory/FailingDataSinkFactory.java
new file mode 100644
index 000000000..c89f7ddbc
--- /dev/null
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/testsink/factory/FailingDataSinkFactory.java
@@ -0,0 +1,224 @@
+/*
+ * 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.testsink.factory;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.exceptions.SchemaEvolveException;
+import org.apache.flink.cdc.common.factories.DataSinkFactory;
+import org.apache.flink.cdc.common.sink.DataSink;
+import org.apache.flink.cdc.common.sink.EventSinkProvider;
+import org.apache.flink.cdc.common.sink.FlinkSinkProvider;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+
+import java.lang.reflect.Field;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/** A test-only sink that fails one sink subtask while schema evolution is in
progress. */
+public class FailingDataSinkFactory implements DataSinkFactory {
+
+ public static final String IDENTIFIER = "schema-evolution-failing-sink";
+ public static final String ORIGINAL_FAILURE_MESSAGE = "Intentional sink
subtask failure";
+
+ private static volatile CountDownLatch metadataApplying = new
CountDownLatch(1);
+ private static volatile CountDownLatch releaseMetadataApplier = new
CountDownLatch(1);
+ private static volatile CountDownLatch sinkFailureTriggered = new
CountDownLatch(1);
+ private static final AtomicBoolean failOnce = new AtomicBoolean();
+
+ public static void reset() {
+ metadataApplying = new CountDownLatch(1);
+ releaseMetadataApplier = new CountDownLatch(1);
+ sinkFailureTriggered = new CountDownLatch(1);
+ failOnce.set(false);
+ }
+
+ public static boolean awaitMetadataApplying(Duration timeout) throws
InterruptedException {
+ return metadataApplying.await(timeout.toMillis(),
TimeUnit.MILLISECONDS);
+ }
+
+ public static boolean awaitSinkFailure(Duration timeout) throws
InterruptedException {
+ return sinkFailureTriggered.await(timeout.toMillis(),
TimeUnit.MILLISECONDS);
+ }
+
+ public static void releaseMetadataApplier() {
+ releaseMetadataApplier.countDown();
+ }
+
+ @Override
+ public DataSink createDataSink(Context context) {
+ return new FailingDataSink();
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Collections.emptySet();
+ }
+
+ private static class FailingDataSink implements DataSink {
+
+ @Override
+ public EventSinkProvider getEventSinkProvider() {
+ return FlinkSinkProvider.of(new FailingSink());
+ }
+
+ @Override
+ public MetadataApplier getMetadataApplier() {
+ return new BlockingMetadataApplier();
+ }
+ }
+
+ private static class FailingSink implements Sink<Event> {
+
+ @Override
+ public SinkWriter<Event> createWriter(InitContext context) {
+ return new FailingSinkWriter(getTaskEnvironment(context),
context.getSubtaskId());
+ }
+
+ private static Environment getTaskEnvironment(InitContext context) {
+ // Sink.InitContext does not expose the task environment needed
for an out-of-band
+ // failure.
+ try {
+ Class<?> initContextBase =
+ Class.forName(
+
"org.apache.flink.streaming.runtime.operators.sink.InitContextBase");
+ Object initContext = unwrapInitContext(context,
initContextBase);
+ Field runtimeContextField =
initContextBase.getDeclaredField("runtimeContext");
+ runtimeContextField.setAccessible(true);
+ Object runtimeContext = runtimeContextField.get(initContext);
+ if (!(runtimeContext instanceof StreamingRuntimeContext)) {
+ throw new IllegalStateException(
+ "Unexpected runtime context: "
+ + (runtimeContext == null
+ ? "null"
+ :
runtimeContext.getClass().getName()));
+ }
+
+ Field taskEnvironment =
+
StreamingRuntimeContext.class.getDeclaredField("taskEnvironment");
+ taskEnvironment.setAccessible(true);
+ return (Environment) taskEnvironment.get(runtimeContext);
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalStateException("Cannot access the task
environment.", e);
+ }
+ }
+
+ private static Object unwrapInitContext(Object context, Class<?>
initContextBase)
+ throws ReflectiveOperationException {
+ Object current = context;
+ while (!initContextBase.isInstance(current)) {
+ String className = current.getClass().getName();
+ String wrappedField;
+ if (className.endsWith("Sink$InitContextWrapper")) {
+ wrappedField = "wrapped";
+ } else if (className.equals(
+
"org.apache.flink.api.connector.sink2.InitContextAdapter")) {
+ wrappedField = "context";
+ } else {
+ throw new NoSuchFieldException("Cannot unwrap init context
" + className);
+ }
+ Field wrapped =
current.getClass().getDeclaredField(wrappedField);
+ wrapped.setAccessible(true);
+ current = wrapped.get(current);
+ }
+ return current;
+ }
+ }
+
+ private static class FailingSinkWriter implements SinkWriter<Event> {
+
+ private final Thread failureThread;
+
+ private FailingSinkWriter(Environment taskEnvironment, int subtaskId) {
+ CountDownLatch metadataApplyingLatch = metadataApplying;
+ failureThread =
+ new Thread(
+ () ->
+ failTaskAfterSchemaEvolutionStarts(
+ taskEnvironment, subtaskId,
metadataApplyingLatch),
+ "schema-evolution-sink-failure");
+ failureThread.setDaemon(true);
+ failureThread.start();
+ }
+
+ private static void failTaskAfterSchemaEvolutionStarts(
+ Environment taskEnvironment, int subtaskId, CountDownLatch
metadataApplyingLatch) {
+ if (subtaskId != 0 || failOnce.get()) {
+ return;
+ }
+
+ try {
+ metadataApplyingLatch.await();
+ if (failOnce.compareAndSet(false, true)) {
+ sinkFailureTriggered.countDown();
+ // Fail the task externally so the sink remains chained
with the schema
+ // operator, independent of the sink's blocked mailbox
thread.
+ taskEnvironment.failExternally(new
RuntimeException(ORIGINAL_FAILURE_MESSAGE));
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ @Override
+ public void write(Event element, Context context) {}
+
+ @Override
+ public void flush(boolean endOfInput) {}
+
+ @Override
+ public void close() {
+ failureThread.interrupt();
+ }
+ }
+
+ private static class BlockingMetadataApplier implements MetadataApplier {
+
+ @Override
+ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent)
+ throws SchemaEvolveException {
+ metadataApplying.countDown();
+ try {
+ releaseMetadataApplier.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new SchemaEvolveException(
+ schemaChangeEvent, "Blocking metadata applier was
interrupted.", e);
+ }
+ }
+ }
+}
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 f5be4c841..03c475c82 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
@@ -19,5 +19,6 @@ 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.testsink.factory.FailingDataSinkFactory
org.apache.flink.cdc.composer.flink.FlinkPipelineComposerTest$TestDataSinkFactory
org.apache.flink.cdc.composer.flink.FlinkPipelineComposerTest$TestDataSourceFactory
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 5dc1ff99e..03ebc68c9 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
@@ -315,7 +315,7 @@ public abstract class SchemaRegistry implements
OperatorCoordinator, Coordinatio
}
@Override
- public final void executionAttemptFailed(
+ public void executionAttemptFailed(
int subTaskId, int attemptNumber, @Nullable Throwable reason) {
if (reason != null) {
failedReasons.put(subTaskId, reason);
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 286f89bfe..18ae29b11 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
@@ -39,6 +39,7 @@ import
org.apache.flink.cdc.runtime.operators.schema.distributed.event.SchemaCha
import org.apache.flink.runtime.operators.coordination.CoordinationRequest;
import org.apache.flink.runtime.operators.coordination.CoordinationResponse;
import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkRuntimeException;
import
org.apache.flink.shaded.guava31.com.google.common.collect.HashBasedTable;
@@ -49,6 +50,8 @@ import
org.apache.flink.shaded.guava31.com.google.common.collect.Table;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
@@ -79,6 +82,13 @@ public class SchemaCoordinator extends SchemaRegistry {
/** Atomic finite state machine to track global schema evolving state. */
private transient AtomicReference<RequestStatus> evolvingStatus;
+ /**
+ * Preserves the first failure observed while a schema evolution is in
progress, so that a later
+ * attempt's failure won't replace the original cause that left the
coordinator in an invalid
+ * state.
+ */
+ private transient AtomicReference<Throwable> firstSchemaEvolutionFailure;
+
/** Request futures from pending schema mappers. */
private transient Map<
Integer, Tuple2<SchemaChangeRequest,
CompletableFuture<CoordinationResponse>>>
@@ -139,6 +149,7 @@ public class SchemaCoordinator extends SchemaRegistry {
"Schema coordinator request was
cancelled by checkpoint reset.")));
}
this.evolvingStatus = new AtomicReference<>(RequestStatus.IDLE);
+ this.firstSchemaEvolutionFailure = new AtomicReference<>();
this.pendingRequests = new ConcurrentHashMap<>();
this.flushedSinkWriters = ConcurrentHashMap.newKeySet();
this.upstreamSchemaTable = HashBasedTable.create();
@@ -174,6 +185,19 @@ public class SchemaCoordinator extends SchemaRegistry {
}
}
+ @Override
+ public void executionAttemptFailed(
+ int subTaskId, int attemptNumber, @Nullable Throwable reason) {
+ if (reason != null) {
+ // Preserve the first failure per subtask to avoid a later attempt
overwriting the
+ // original cause that left the coordinator in an invalid state.
+ failedReasons.putIfAbsent(subTaskId, reason);
+ if (hasOngoingSchemaEvolution()) {
+ firstSchemaEvolutionFailure.compareAndSet(null, reason);
+ }
+ }
+ }
+
// --------------------------
// Checkpoint related methods
// --------------------------
@@ -283,10 +307,12 @@ public class SchemaCoordinator extends SchemaRegistry {
pendingRequests.put(request.getSinkSubTaskId(), Tuple2.of(request,
responseFuture));
if (pendingRequests.size() == 1) {
- Preconditions.checkState(
+ safeCheckState(
evolvingStatus.compareAndSet(
RequestStatus.IDLE,
RequestStatus.WAITING_FOR_FLUSH),
- "Unexpected evolving status: " + evolvingStatus.get());
+ request.getSinkSubTaskId());
+ failedReasons.clear();
+ firstSchemaEvolutionFailure.set(null);
LOG.info(
"Received the very-first schema change request {}.
Switching from IDLE to WAITING_FOR_FLUSH.",
request);
@@ -294,10 +320,10 @@ public class SchemaCoordinator extends SchemaRegistry {
// No else if, since currentParallelism might be == 1
if (pendingRequests.size() == currentParallelism) {
- Preconditions.checkState(
+ safeCheckState(
evolvingStatus.compareAndSet(
RequestStatus.WAITING_FOR_FLUSH,
RequestStatus.EVOLVING),
- "Unexpected evolving status: " + evolvingStatus.get());
+ request.getSinkSubTaskId());
LOG.info(
"Received the last required schema change request {}.
Switching from WAITING_FOR_FLUSH to EVOLVING.",
request);
@@ -318,6 +344,31 @@ public class SchemaCoordinator extends SchemaRegistry {
}
}
+ private void safeCheckState(boolean condition, int sinkSubTaskId) throws
Exception {
+ if (condition) {
+ return;
+ }
+
+ IllegalStateException stateError =
+ new IllegalStateException("Unexpected evolving status: " +
evolvingStatus.get());
+ Throwable originalFailure = firstSchemaEvolutionFailure.get();
+ if (originalFailure == null) {
+ originalFailure = failedReasons.get(sinkSubTaskId);
+ }
+ if (originalFailure == null && !failedReasons.isEmpty()) {
+ originalFailure = failedReasons.values().iterator().next();
+ }
+ if (originalFailure == null) {
+ throw stateError;
+ }
+
+ originalFailure.addSuppressed(stateError);
+ LOG.warn(
+ "Preserving the original subtask failure instead of reporting
a secondary schema evolution state error.",
+ originalFailure);
+ ExceptionUtils.rethrowException(originalFailure);
+ }
+
/**
* Tries to apply schema change event {@code schemaChangeEvent} to the
combination of {@code
* tableId} and {@code sourcePartition}. Returns {@code true} if schema
got changed, or {@code
@@ -513,4 +564,16 @@ public class SchemaCoordinator extends SchemaRegistry {
public void emplaceOriginalSchema(TableId tableId, Integer subTaskId,
Schema schema) {
upstreamSchemaTable.put(tableId, subTaskId, schema);
}
+
+ private boolean hasOngoingSchemaEvolution() {
+ return evolvingStatus != null
+ && (evolvingStatus.get() != RequestStatus.IDLE
+ || (pendingRequests != null &&
!pendingRequests.isEmpty())
+ || (flushedSinkWriters != null &&
!flushedSinkWriters.isEmpty()));
+ }
+
+ @VisibleForTesting
+ boolean isSchemaEvolutionInProgress() {
+ return evolvingStatus != null && evolvingStatus.get() ==
RequestStatus.EVOLVING;
+ }
}
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
index 35d92e176..63ecd3f90 100644
---
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
@@ -35,10 +35,13 @@ import
org.apache.flink.cdc.runtime.operators.schema.common.event.GetEvolvedSche
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.cdc.runtime.testutils.schema.CollectingMetadataApplier;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.coordination.CoordinationResponse;
+import
org.apache.flink.runtime.operators.coordination.MockOperatorCoordinatorContext;
import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
+import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -61,6 +64,7 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BooleanSupplier;
import static
org.apache.flink.cdc.runtime.operators.schema.common.CoordinationResponseUtils.unwrap;
import static org.assertj.core.api.Assertions.assertThat;
@@ -284,6 +288,196 @@ class SchemaCoordinatorTest {
}
}
+ @Test
+ void testIgnoreSubtaskResetWithoutOngoingSchemaEvolution() throws
Exception {
+ MockedOperatorCoordinatorContext context = createContext();
+ SchemaCoordinator coordinator = createCoordinator(context);
+ coordinator.start();
+
+ try {
+ coordinator.executionAttemptFailed(
+ 0, 0, new RuntimeException("failure outside schema
evolution"));
+ coordinator.subtaskReset(0, 123L);
+
+ Assertions.assertThat(context.isJobFailed()).isFalse();
+ } finally {
+ coordinator.close();
+ }
+ }
+
+ @Test
+ void testDoNotEscalateSubtaskResetWhenFailureReasonIsMissing() throws
Exception {
+ MockedOperatorCoordinatorContext context = createContext();
+ SchemaCoordinator coordinator = createCoordinator(context);
+ coordinator.start();
+
+ try {
+ CompletableFuture<CoordinationResponse> requestFuture =
+
coordinator.handleCoordinationRequest(createSchemaChangeRequest());
+ waitUntil(coordinator::isSchemaEvolutionInProgress);
+
+ coordinator.subtaskReset(0, 123L);
+ coordinator.subtaskReset(0, 124L);
+
+ Assertions.assertThat(context.isJobFailed()).isFalse();
+ Assertions.assertThat(requestFuture).isNotDone();
+ } finally {
+ coordinator.close();
+ }
+ }
+
+ @Test
+ void testPreserveFailureFromAnotherSubtaskWithoutEscalatingReset() throws
Exception {
+ MockOperatorCoordinatorContext context =
+ new MockOperatorCoordinatorContext(
+ new OperatorID(), 2,
Thread.currentThread().getContextClassLoader());
+ SchemaCoordinator coordinator = createCoordinator(context);
+ coordinator.start();
+
+ try {
+ CompletableFuture<CoordinationResponse> requestFromSubtaskZero =
+
coordinator.handleCoordinationRequest(createSchemaChangeRequest(0));
+
coordinator.handleCoordinationRequest(createSchemaChangeRequest(1));
+ waitUntil(coordinator::isSchemaEvolutionInProgress);
+
+ RuntimeException originalFailure = new RuntimeException("other
subtask failure");
+ coordinator.executionAttemptFailed(1, 0, originalFailure);
+ coordinator.subtaskReset(1, 123L);
+
+ Assertions.assertThat(context.isJobFailed()).isFalse();
+ coordinator.executionAttemptReady(1, 1, null);
+ CompletableFuture<CoordinationResponse> retriedRequest =
+
coordinator.handleCoordinationRequest(createSchemaChangeRequest(0));
+
+ waitUntil(context::isJobFailed);
+
Assertions.assertThat(context.getJobFailureReason()).isSameAs(originalFailure);
+ Assertions.assertThatThrownBy(() -> retriedRequest.get(5,
TimeUnit.SECONDS))
+ .hasCause(originalFailure);
+ assertUnexpectedEvolvingStatusSuppressed(originalFailure);
+ Assertions.assertThat(requestFromSubtaskZero).isNotDone();
+ } finally {
+ coordinator.close();
+ }
+ }
+
+ @Test
+ void testPreserveOriginalFailureDuringPartialFailover() throws Exception {
+ MockedOperatorCoordinatorContext context = createContext();
+ SchemaCoordinator coordinator = createCoordinator(context);
+ coordinator.start();
+
+ try {
+ SchemaChangeRequest request = createSchemaChangeRequest();
+
+ CompletableFuture<CoordinationResponse> firstRequest =
+ coordinator.handleCoordinationRequest(request);
+ waitUntil(coordinator::isSchemaEvolutionInProgress);
+
+ RuntimeException originalFailure = new RuntimeException("original
task failure");
+ coordinator.executionAttemptFailed(0, 0, originalFailure);
+ coordinator.subtaskReset(0, 123L);
+
+ Assertions.assertThat(context.isJobFailed()).isFalse();
+ coordinator.executionAttemptReady(0, 1, null);
+ CompletableFuture<CoordinationResponse> retriedRequest =
+ coordinator.handleCoordinationRequest(request);
+
+ waitUntil(context::isJobFailed);
+
Assertions.assertThat(context.getFailureCause()).isSameAs(originalFailure);
+ Assertions.assertThatThrownBy(() -> retriedRequest.get(5,
TimeUnit.SECONDS))
+ .hasCause(originalFailure);
+ assertUnexpectedEvolvingStatusSuppressed(originalFailure);
+ Assertions.assertThat(firstRequest).isNotDone();
+ } finally {
+ coordinator.close();
+ }
+ }
+
+ @Test
+ void testPreserveFirstFailureAcrossMultipleAttempts() throws Exception {
+ MockedOperatorCoordinatorContext context = createContext();
+ SchemaCoordinator coordinator = createCoordinator(context);
+ coordinator.start();
+
+ try {
+ SchemaChangeRequest request = createSchemaChangeRequest();
+
+ CompletableFuture<CoordinationResponse> firstRequest =
+ coordinator.handleCoordinationRequest(request);
+ waitUntil(coordinator::isSchemaEvolutionInProgress);
+
+ RuntimeException firstFailure = new RuntimeException("first
attempt failure");
+ coordinator.executionAttemptFailed(0, 0, firstFailure);
+ coordinator.subtaskReset(0, 123L);
+
+ RuntimeException secondFailure = new RuntimeException("second
attempt failure");
+ coordinator.executionAttemptFailed(0, 1, secondFailure);
+ coordinator.subtaskReset(0, 124L);
+
+ coordinator.executionAttemptReady(0, 2, null);
+ CompletableFuture<CoordinationResponse> retriedRequest =
+ coordinator.handleCoordinationRequest(request);
+
+ waitUntil(context::isJobFailed);
+
Assertions.assertThat(context.getFailureCause()).isSameAs(firstFailure);
+ Assertions.assertThatThrownBy(() -> retriedRequest.get(5,
TimeUnit.SECONDS))
+ .hasCause(firstFailure);
+ assertUnexpectedEvolvingStatusSuppressed(firstFailure);
+ Assertions.assertThat(firstRequest).isNotDone();
+ } finally {
+ coordinator.close();
+ }
+ }
+
+ @Test
+ void testRetainStateErrorWhenOriginalFailureIsMissing() throws Exception {
+ MockedOperatorCoordinatorContext context = createContext();
+ SchemaCoordinator coordinator = createCoordinator(context);
+ coordinator.start();
+
+ try {
+ SchemaChangeRequest request = createSchemaChangeRequest();
+ coordinator.handleCoordinationRequest(request);
+ waitUntil(coordinator::isSchemaEvolutionInProgress);
+
+ coordinator.handleCoordinationRequest(request);
+
+ waitUntil(context::isJobFailed);
+ Assertions.assertThat(context.getFailureCause())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Unexpected evolving status: EVOLVING");
+ } finally {
+ coordinator.close();
+ }
+ }
+
+ @Test
+ void testClearStaleFailureWhenStartingNewSchemaEvolution() throws
Exception {
+ MockedOperatorCoordinatorContext context = createContext();
+ SchemaCoordinator coordinator = createCoordinator(context);
+ coordinator.start();
+
+ try {
+ RuntimeException staleFailure = new RuntimeException("stale task
failure");
+ coordinator.executionAttemptFailed(0, 0, staleFailure);
+
+ SchemaChangeRequest request = createSchemaChangeRequest();
+ coordinator.handleCoordinationRequest(request);
+ waitUntil(coordinator::isSchemaEvolutionInProgress);
+
+ coordinator.handleCoordinationRequest(request);
+
+ waitUntil(context::isJobFailed);
+ Assertions.assertThat(context.getFailureCause())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Unexpected evolving status: EVOLVING")
+ .isNotSameAs(staleFailure);
+ Assertions.assertThat(staleFailure.getSuppressed()).isEmpty();
+ } finally {
+ coordinator.close();
+ }
+ }
+
private static SchemaCoordinator createCoordinator(
FailOnceMetadataApplier metadataApplier, Duration rpcTimeout) {
return createCoordinator(
@@ -327,6 +521,54 @@ class SchemaCoordinatorTest {
return response;
}
+ private static MockedOperatorCoordinatorContext createContext() {
+ return new MockedOperatorCoordinatorContext(
+ new OperatorID(),
Thread.currentThread().getContextClassLoader());
+ }
+
+ private static SchemaCoordinator
createCoordinator(OperatorCoordinator.Context context) {
+ return new SchemaCoordinator(
+ "SchemaCoordinator",
+ context,
+ Executors.newSingleThreadExecutor(),
+ new CollectingMetadataApplier(Duration.ZERO),
+ Collections.emptyList(),
+ RouteMode.ALL_MATCH,
+ SchemaChangeBehavior.LENIENT,
+ Duration.ofMinutes(1));
+ }
+
+ private static SchemaChangeRequest createSchemaChangeRequest() {
+ return createSchemaChangeRequest(0);
+ }
+
+ private static SchemaChangeRequest createSchemaChangeRequest(int
sinkSubTaskId) {
+ TableId tableId = TableId.tableId("inventory", "products");
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .primaryKey("id")
+ .build();
+ return new SchemaChangeRequest(0, sinkSubTaskId, new
CreateTableEvent(tableId, schema));
+ }
+
+ private static void assertUnexpectedEvolvingStatusSuppressed(Throwable
originalFailure) {
+ Assertions.assertThat(originalFailure.getSuppressed()).hasSize(1);
+ Assertions.assertThat(originalFailure.getSuppressed()[0])
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Unexpected evolving status: EVOLVING");
+ }
+
+ private static void waitUntil(BooleanSupplier condition) throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (!condition.getAsBoolean()) {
+ if (System.nanoTime() > deadline) {
+ throw new AssertionError("Condition was not met within the
timeout.");
+ }
+ Thread.sleep(10);
+ }
+ }
+
private static final class FailOnceMetadataApplier implements
MetadataApplier {
private final AtomicBoolean failNextAdd;
private final CountDownLatch failedAdd = new CountDownLatch(1);
