DanielLeens commented on PR #11682: URL: https://github.com/apache/seatunnel/pull/11682#issuecomment-6368679481
I prepared an apply-ready follow-up patch for the remaining source-level points on head `ddb5d0ab64d91fc3883ef94e57741223d3a17718`. It preserves the optional job-id lookup semantics and the existing public constructors. The patch adds structural-mismatch DEBUG diagnostics, tests event job-id enrichment with real direct/wrapped `SourceCoordinatorContext` instances, covers a wrapped context whose failure handler is null, and verifies that lifecycle events use the listener supplied through the shared four-argument enumerator context. The lifecycle test fails if that constructor recreates the context and loses the original event listener; the positive job-id tests fail if the reflection/unwrap path stops enriching events. Please also scope the PR description to optional job-id enrichment/context reuse. The discussion in [#10193](https://github.com/apache/seatunnel/issues/10193#issuecomment-3691171772) treats the reflection exception as a warning, and the reporter's later [Kafka transaction investigation](https://github.com/apache/seatunnel/issues/10193#issuecomment-3698140367) led to [#10013](https://github.com/apache/seatunnel/issues/10193#issuecomment-3708995025). This patch does not establish that the separate checkpoint-restore failure is fixed. To apply, save the diff below as `pr11682-context-diagnostics.patch` on the current PR branch, confirm `git rev-parse HEAD` equals `ddb5d0ab64d91fc3883ef94e57741223d3a17718`, then run: ```bash git apply --check pr11682-context-diagnostics.patch git apply pr11682-context-diagnostics.patch ``` Validation performed: scoped `./mvnw spotless:apply -pl seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common -nsu -Dmaven.gitcommitid.skip=true`, `git diff --check`, and patch applicability against the exact head all passed. Compilation, unit tests, and Flink runtime/restore verification have not run on this patch. After applying/pushing, the branch owner must enable its workflows and obtain fresh CI evidence; the existing Build gate reports workflow-run detection failure. The new tests cover the common module's declared Flink 1.15.3 dependency and reflection-shaped restore fixtures; they are not a Flink 1.16/1.17/1.18 runtime matrix. <details> <summary>Exact-head contributor patch</summary> ```diff diff --git a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/source/FlinkSourceSplitEnumeratorContext.java b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/source/FlinkSourceSplitEnumeratorContext.java index baddc01d8c8..e2ab52a8b16 100644 --- a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/source/FlinkSourceSplitEnumeratorContext.java +++ b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/main/java/org/apache/seatunnel/translation/flink/source/FlinkSourceSplitEnumeratorContext.java @@ -106,8 +106,8 @@ public class FlinkSourceSplitEnumeratorContext<SplitT extends SourceSplit> } /** - * Best-effort Flink job id resolution for event enrichment. Failure must never abort - * checkpoint/savepoint restore (see issue #10193). + * Best-effort Flink job id resolution for event enrichment. This metadata lookup must not + * prevent enumerator construction; it does not determine checkpoint/savepoint restore success. */ private static String getFlinkJobId(SplitEnumeratorContext enumContext) { try { @@ -129,12 +129,16 @@ public class FlinkSourceSplitEnumeratorContext<SplitT extends SourceSplit> */ private static String getJobIdForV15(SplitEnumeratorContext enumContext) throws Exception { if (!(enumContext instanceof SourceCoordinatorContext)) { + log.debug("Flink job id unavailable: context is not a SourceCoordinatorContext."); return null; } SourceCoordinatorContext coordinatorContext = (SourceCoordinatorContext) enumContext; Field operatorCoordinatorContextField = findDeclaredField(coordinatorContext.getClass(), "operatorCoordinatorContext"); if (operatorCoordinatorContextField == null) { + log.debug( + "Flink job id unavailable: {} has no operatorCoordinatorContext field.", + coordinatorContext.getClass().getName()); return null; } operatorCoordinatorContextField.setAccessible(true); @@ -142,6 +146,7 @@ public class FlinkSourceSplitEnumeratorContext<SplitT extends SourceSplit> (OperatorCoordinator.Context) operatorCoordinatorContextField.get(coordinatorContext); if (operatorCoordinatorContext == null) { + log.debug("Flink job id unavailable: operator coordinator context is null."); return null; } @@ -161,6 +166,7 @@ public class FlinkSourceSplitEnumeratorContext<SplitT extends SourceSplit> (OperatorCoordinator.Context) nestedContextField.get(operatorCoordinatorContext); if (operatorCoordinatorContext == null) { + log.debug("Flink job id unavailable: nested restore coordinator context is null."); return null; } } @@ -168,6 +174,9 @@ public class FlinkSourceSplitEnumeratorContext<SplitT extends SourceSplit> Field globalFailureHandlerField = findDeclaredField(operatorCoordinatorContext.getClass(), "globalFailureHandler"); if (globalFailureHandlerField == null) { + log.debug( + "Flink job id unavailable: {} has no globalFailureHandler field.", + operatorCoordinatorContext.getClass().getName()); return null; } globalFailureHandlerField.setAccessible(true); @@ -182,11 +191,15 @@ public class FlinkSourceSplitEnumeratorContext<SplitT extends SourceSplit> Field schedulerField = findDeclaredField(globalFailureHandler.getClass(), "arg$1"); if (schedulerField == null) { + log.debug( + "Flink job id unavailable: failure handler {} has no scheduler field arg$1.", + globalFailureHandler.getClass().getName()); return null; } schedulerField.setAccessible(true); Object scheduler = schedulerField.get(globalFailureHandler); if (!(scheduler instanceof SchedulerBase)) { + log.debug("Flink job id unavailable: reflected scheduler is not a SchedulerBase."); return null; } return ((SchedulerBase) scheduler).getExecutionGraph().getJobID().toString(); diff --git a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/source/FlinkSourceEnumeratorTest.java b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/source/FlinkSourceEnumeratorTest.java index 83ca41af734..bb789b507a0 100644 --- a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/source/FlinkSourceEnumeratorTest.java +++ b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/source/FlinkSourceEnumeratorTest.java @@ -17,8 +17,13 @@ package org.apache.seatunnel.translation.flink.source; +import org.apache.seatunnel.api.event.DefaultEventProcessor; +import org.apache.seatunnel.api.event.Event; +import org.apache.seatunnel.api.event.EventListener; import org.apache.seatunnel.api.source.SourceSplit; import org.apache.seatunnel.api.source.SourceSplitEnumerator; +import org.apache.seatunnel.api.source.event.EnumeratorCloseEvent; +import org.apache.seatunnel.api.source.event.EnumeratorOpenEvent; import org.apache.flink.api.connector.source.SplitEnumeratorContext; @@ -27,6 +32,9 @@ import org.junit.jupiter.api.Test; import org.mockito.Mockito; import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; @@ -150,4 +158,41 @@ class FlinkSourceEnumeratorTest { executorService.shutdownNow(); } } + + /** + * Lifecycle events must use the supplied context's listener, preserving its job identity. + * Recreating a context in the four-argument constructor loses these enriched events. + */ + @Test + void testLifecycleEventsUseSharedContext() throws Exception { + SourceSplitEnumerator<DummySplit, Serializable> sourceSplitEnumerator = + Mockito.mock(SourceSplitEnumerator.class); + SplitEnumeratorContext<SplitWrapper<DummySplit>> enumeratorContext = + Mockito.mock(SplitEnumeratorContext.class); + List<Event> events = new ArrayList<>(); + EventListener listener = + new DefaultEventProcessor("restored-job", Collections.singletonList(events::add)); + SourceSplitEnumerator.Context<DummySplit> sharedContext = + new FlinkSourceSplitEnumeratorContext<DummySplit>(enumeratorContext) { + @Override + public EventListener getEventListener() { + return listener; + } + }; + FlinkSourceEnumerator<DummySplit, Serializable> enumerator = + new FlinkSourceEnumerator<>( + sourceSplitEnumerator, + enumeratorContext, + sharedContext, + ConcurrentHashMap.newKeySet()); + + enumerator.start(); + enumerator.close(); + + Assertions.assertEquals(2, events.size()); + Assertions.assertEquals(EnumeratorOpenEvent.class, events.get(0).getClass()); + Assertions.assertEquals(EnumeratorCloseEvent.class, events.get(1).getClass()); + Assertions.assertEquals("restored-job", events.get(0).getJobId()); + Assertions.assertEquals("restored-job", events.get(1).getJobId()); + } } diff --git a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/source/FlinkSourceSplitEnumeratorContextTest.java b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/source/FlinkSourceSplitEnumeratorContextTest.java index 5b7f9e805f0..a2fa239a7c5 100644 --- a/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/source/FlinkSourceSplitEnumeratorContextTest.java +++ b/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/source/FlinkSourceSplitEnumeratorContextTest.java @@ -18,8 +18,18 @@ package org.apache.seatunnel.translation.flink.source; import org.apache.seatunnel.api.source.SourceSplit; +import org.apache.seatunnel.api.source.event.EnumeratorOpenEvent; +import org.apache.flink.api.common.JobID; import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.core.io.SimpleVersionedSerializer; +import org.apache.flink.runtime.executiongraph.ExecutionGraph; +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.operators.coordination.CoordinatorStore; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.runtime.scheduler.SchedulerBase; +import org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext; +import org.apache.flink.runtime.source.coordinator.SourceCoordinatorProvider.CoordinatorExecutorThreadFactory; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -29,8 +39,8 @@ import java.util.Collections; import java.util.concurrent.atomic.AtomicBoolean; /** - * Regression for issue #10193: resolving Flink job id must not abort enumerator construction during - * checkpoint/savepoint restore when Flink internals are unavailable. + * Covers job-id event enrichment through direct and wrapped Flink coordinator contexts, and + * best-effort context construction when the optional metadata cannot be resolved. */ class FlinkSourceSplitEnumeratorContextTest { @@ -66,4 +76,119 @@ class FlinkSourceSplitEnumeratorContextTest { Assertions.assertTrue(signaled.get()); Mockito.verify(enumContext).signalNoMoreSplits(0); } + + @Test + void resolveJobIdFromCoordinatorContext() throws Exception { + JobID jobId = new JobID(); + assertEventJobId( + new FailureHandlerContext(schedulerFailureHandler(jobId)), jobId.toString()); + } + + @Test + void resolveJobIdFromWrappedRestoreContext() throws Exception { + JobID jobId = new JobID(); + assertEventJobId( + new WrappedContext(new FailureHandlerContext(schedulerFailureHandler(jobId))), + jobId.toString()); + } + + @Test + void wrappedRestoreContextWithNullFailureHandlerPreservesSourceContext() throws Exception { + assertEventJobId(new WrappedContext(new FailureHandlerContext(null)), null); + } + + private SchedulerFailureHandler schedulerFailureHandler(JobID jobId) { + ExecutionGraph executionGraph = Mockito.mock(ExecutionGraph.class); + Mockito.when(executionGraph.getJobID()).thenReturn(jobId); + SchedulerBase scheduler = Mockito.mock(SchedulerBase.class); + Mockito.when(scheduler.getExecutionGraph()).thenReturn(executionGraph); + return new SchedulerFailureHandler(scheduler); + } + + private void assertEventJobId(OperatorCoordinator.Context operatorContext, String expectedJobId) + throws Exception { + try (SourceCoordinatorContext<SplitWrapper<DummySplit>> flinkContext = + new SourceCoordinatorContext<>( + Mockito.mock(CoordinatorExecutorThreadFactory.class), + 1, + operatorContext, + Mockito.mock(SimpleVersionedSerializer.class))) { + FlinkSourceSplitEnumeratorContext<DummySplit> context = + new FlinkSourceSplitEnumeratorContext<>(flinkContext); + EnumeratorOpenEvent event = new EnumeratorOpenEvent(); + + context.getEventListener().onEvent(event); + + Assertions.assertEquals(expectedJobId, event.getJobId()); + Assertions.assertEquals(1, context.currentParallelism()); + Assertions.assertEquals(Collections.emptySet(), context.registeredReaders()); + } + } + + /** + * Supplies stable coordinator metadata without starting a Flink job. Unexpected job failure + * callbacks fail the test instead of being ignored by a coordinator mock. + */ + private abstract static class TestOperatorContext implements OperatorCoordinator.Context { + @Override + public OperatorID getOperatorId() { + return new OperatorID(); + } + + @Override + public void failJob(Throwable cause) { + throw new AssertionError("Job-id enrichment must not fail the job", cause); + } + + @Override + public int currentParallelism() { + return 1; + } + + @Override + public ClassLoader getUserCodeClassloader() { + return getClass().getClassLoader(); + } + + @Override + public CoordinatorStore getCoordinatorStore() { + return null; + } + } + + /** + * Models the wrapper field shape used by Flink's reset coordinator. The real resolver must + * unwrap this context before looking for the failure handler. + */ + private static final class WrappedContext extends TestOperatorContext { + private final OperatorCoordinator.Context context; + + private WrappedContext(OperatorCoordinator.Context context) { + this.context = context; + } + } + + /** + * Models the lazily initialized coordinator's optional failure handler, including the null + * value exposed before the scheduler has completed restore initialization. + */ + private static final class FailureHandlerContext extends TestOperatorContext { + private final Object globalFailureHandler; + + private FailureHandlerContext(Object globalFailureHandler) { + this.globalFailureHandler = globalFailureHandler; + } + } + + /** + * Models the scheduler captured by Flink's failure-handler closure. The field name matches the + * existing production reflection contract rather than a test-only accessor. + */ + private static final class SchedulerFailureHandler { + private final SchedulerBase arg$1; + + private SchedulerFailureHandler(SchedulerBase scheduler) { + this.arg$1 = scheduler; + } + } } ``` </details> <!-- contributor-patch:11682:ddb5d0ab64d91fc3883ef94e57741223d3a17718 --> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
