This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 823d9d509f2 [FLINK-39776][runtime][core] Expose JobInfo on
SourceOperators
823d9d509f2 is described below
commit 823d9d509f2f0b83394dfa7958ff4fea44b94e41
Author: Aleksandr Savonin <[email protected]>
AuthorDate: Wed May 27 21:08:12 2026 +0200
[FLINK-39776][runtime][core] Expose JobInfo on SourceOperators
---
.../source/hybrid/HybridSourceSplitEnumerator.java | 6 ++
.../connector/source/RichSourceReaderContext.java | 6 ++
.../api/connector/source/SourceReaderContext.java | 12 +++
.../connector/source/SplitEnumeratorContext.java | 11 +++
.../source/mocks/MockSplitEnumeratorContext.java | 10 +++
.../InternalExecutionGraphAccessor.java | 2 +
.../coordination/OperatorCoordinator.java | 4 +
.../coordination/OperatorCoordinatorHolder.java | 23 ++++--
.../RecreateOnResetOperatorCoordinator.java | 6 ++
.../coordinator/SourceCoordinatorContext.java | 6 ++
.../source/RichSourceReaderContextTest.java | 89 ++++++++++++++++++++++
.../MockOperatorCoordinatorContext.java | 8 ++
.../OperatorCoordinatorHolderTest.java | 2 +
.../runtime/scheduler/adaptive/ExecutingTest.java | 5 ++
.../coordinator/SourceCoordinatorContextTest.java | 11 +++
.../source/reader/TestingReaderContext.java | 10 +++
.../reader/TestingSplitEnumeratorContext.java | 10 +++
17 files changed, 216 insertions(+), 5 deletions(-)
diff --git
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitEnumerator.java
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitEnumerator.java
index 48bec9753c5..3e316c4f548 100644
---
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitEnumerator.java
+++
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitEnumerator.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.base.source.hybrid;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.connector.source.ReaderInfo;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.api.connector.source.SourceEvent;
@@ -340,6 +341,11 @@ public class HybridSourceSplitEnumerator
this.sourceSize = sourceSize;
}
+ @Override
+ public JobInfo getJobInfo() {
+ return realContext.getJobInfo();
+ }
+
@Override
public SplitEnumeratorMetricGroup metricGroup() {
return realContext.metricGroup();
diff --git
a/flink-core/src/main/java/org/apache/flink/api/connector/source/RichSourceReaderContext.java
b/flink-core/src/main/java/org/apache/flink/api/connector/source/RichSourceReaderContext.java
index c398cf80c8a..51ba524673b 100644
---
a/flink-core/src/main/java/org/apache/flink/api/connector/source/RichSourceReaderContext.java
+++
b/flink-core/src/main/java/org/apache/flink/api/connector/source/RichSourceReaderContext.java
@@ -19,6 +19,7 @@
package org.apache.flink.api.connector.source;
import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.util.Preconditions;
@@ -39,4 +40,9 @@ public abstract class RichSourceReaderContext implements
SourceReaderContext {
Preconditions.checkNotNull(runtimeContext, "The runtime context must
not be null.");
return runtimeContext;
}
+
+ @Override
+ public JobInfo getJobInfo() {
+ return getRuntimeContext().getJobInfo();
+ }
}
diff --git
a/flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReaderContext.java
b/flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReaderContext.java
index 0973350de81..be0f85d9aba 100644
---
a/flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReaderContext.java
+++
b/flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReaderContext.java
@@ -19,6 +19,8 @@
package org.apache.flink.api.connector.source;
import org.apache.flink.annotation.Public;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.common.watermark.Watermark;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.groups.SourceReaderMetricGroup;
@@ -28,6 +30,16 @@ import org.apache.flink.util.UserCodeClassLoader;
@Public
public interface SourceReaderContext {
+ /**
+ * Get the meta information of the current job.
+ *
+ * @return the job meta information.
+ */
+ @PublicEvolving
+ default JobInfo getJobInfo() {
+ throw new UnsupportedOperationException();
+ }
+
/**
* @return The metric group this source belongs to.
*/
diff --git
a/flink-core/src/main/java/org/apache/flink/api/connector/source/SplitEnumeratorContext.java
b/flink-core/src/main/java/org/apache/flink/api/connector/source/SplitEnumeratorContext.java
index 6c3637b2772..5a0b99020fd 100644
---
a/flink-core/src/main/java/org/apache/flink/api/connector/source/SplitEnumeratorContext.java
+++
b/flink-core/src/main/java/org/apache/flink/api/connector/source/SplitEnumeratorContext.java
@@ -20,6 +20,7 @@ package org.apache.flink.api.connector.source;
import org.apache.flink.annotation.Public;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup;
import java.util.Map;
@@ -41,6 +42,16 @@ import java.util.function.BiConsumer;
@Public
public interface SplitEnumeratorContext<SplitT extends SourceSplit> {
+ /**
+ * Get the meta information of the current job.
+ *
+ * @return the job meta information.
+ */
+ @PublicEvolving
+ default JobInfo getJobInfo() {
+ throw new UnsupportedOperationException();
+ }
+
SplitEnumeratorMetricGroup metricGroup();
/**
diff --git
a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSplitEnumeratorContext.java
b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSplitEnumeratorContext.java
index a09cb28869e..385cad14a90 100644
---
a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSplitEnumeratorContext.java
+++
b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSplitEnumeratorContext.java
@@ -18,6 +18,9 @@ limitations under the License.
package org.apache.flink.api.connector.source.mocks;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.JobInfoImpl;
import org.apache.flink.api.connector.source.ReaderInfo;
import org.apache.flink.api.connector.source.SourceEvent;
import org.apache.flink.api.connector.source.SourceSplit;
@@ -66,6 +69,8 @@ public class MockSplitEnumeratorContext<SplitT extends
SourceSplit>
private final int parallelism;
+ private final JobInfo jobInfo = new JobInfoImpl(new JobID(), "mock-job");
+
public MockSplitEnumeratorContext(int parallelism) {
this.sentSourceEvent = new HashMap<>();
this.registeredReaders = new ConcurrentHashMap<>();
@@ -83,6 +88,11 @@ public class MockSplitEnumeratorContext<SplitT extends
SourceSplit>
this.subtaskHasNoMoreSplits = new boolean[parallelism];
}
+ @Override
+ public JobInfo getJobInfo() {
+ return jobInfo;
+ }
+
@Override
public SplitEnumeratorMetricGroup metricGroup() {
return UnregisteredMetricsGroup.createSplitEnumeratorMetricGroup();
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/InternalExecutionGraphAccessor.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/InternalExecutionGraphAccessor.java
index 364aa4008bd..6802b052596 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/InternalExecutionGraphAccessor.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/InternalExecutionGraphAccessor.java
@@ -48,6 +48,8 @@ public interface InternalExecutionGraphAccessor {
JobID getJobID();
+ String getJobName();
+
BlobWriter getBlobWriter();
/**
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinator.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinator.java
index 2f07f9a639a..ea0798dc5f9 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinator.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinator.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.operators.coordination;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.common.state.CheckpointListener;
import org.apache.flink.metrics.groups.OperatorCoordinatorMetricGroup;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
@@ -261,6 +262,9 @@ public interface OperatorCoordinator extends
CheckpointListener, AutoCloseable {
/** Gets the {@link JobID} of the job to which the coordinator
belongs. */
JobID getJobID();
+ /** Gets the {@link JobInfo} of the job to which the coordinator
belongs. */
+ JobInfo getJobInfo();
+
/** Gets the ID of the operator to which the coordinator belongs. */
OperatorID getOperatorId();
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolder.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolder.java
index 99ce26f0d34..192aacd0f56 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolder.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolder.java
@@ -21,6 +21,8 @@ package org.apache.flink.runtime.operators.coordination;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.JobInfoImpl;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.OperatorCoordinatorMetricGroup;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
@@ -500,6 +502,10 @@ public class OperatorCoordinatorHolder
final SubtaskAccess.SubtaskAccessFactory taskAccesses =
new
ExecutionSubtaskAccess.ExecutionJobVertexSubtaskAccess(jobVertex, opId);
+ final JobInfo jobInfo =
+ new JobInfoImpl(
+ jobVertex.getGraph().getJobID(),
jobVertex.getGraph().getJobName());
+
return create(
opId,
provider,
@@ -511,6 +517,7 @@ public class OperatorCoordinatorHolder
taskAccesses,
supportsConcurrentExecutionAttempts,
taskInformation,
+ jobInfo,
metricGroup);
}
}
@@ -527,6 +534,7 @@ public class OperatorCoordinatorHolder
final SubtaskAccess.SubtaskAccessFactory taskAccesses,
final boolean supportsConcurrentExecutionAttempts,
final TaskInformation taskInformation,
+ final JobInfo jobInfo,
final JobManagerJobMetricGroup jobManagerJobMetricGroup)
throws Exception {
final MetricGroup parentMetricGroup =
@@ -537,7 +545,7 @@ public class OperatorCoordinatorHolder
operatorName);
final LazyInitializedCoordinatorContext context =
new LazyInitializedCoordinatorContext(
- jobManagerJobMetricGroup.jobId(),
+ jobInfo,
opId,
operatorName,
userCodeClassLoader,
@@ -576,7 +584,7 @@ public class OperatorCoordinatorHolder
private static final Logger LOG =
LoggerFactory.getLogger(LazyInitializedCoordinatorContext.class);
- private final JobID jobID;
+ private final JobInfo jobInfo;
private final OperatorID operatorId;
private final String operatorName;
private final ClassLoader userCodeClassLoader;
@@ -592,7 +600,7 @@ public class OperatorCoordinatorHolder
private volatile boolean failed;
public LazyInitializedCoordinatorContext(
- JobID jobID,
+ JobInfo jobInfo,
final OperatorID operatorId,
final String operatorName,
final ClassLoader userCodeClassLoader,
@@ -600,7 +608,7 @@ public class OperatorCoordinatorHolder
final CoordinatorStore coordinatorStore,
final boolean supportsConcurrentExecutionAttempts,
final OperatorCoordinatorMetricGroup metricGroup) {
- this.jobID = jobID;
+ this.jobInfo = checkNotNull(jobInfo);
this.operatorId = checkNotNull(operatorId);
this.operatorName = checkNotNull(operatorName);
this.userCodeClassLoader = checkNotNull(userCodeClassLoader);
@@ -641,7 +649,12 @@ public class OperatorCoordinatorHolder
@Override
public JobID getJobID() {
- return jobID;
+ return jobInfo.getJobId();
+ }
+
+ @Override
+ public JobInfo getJobInfo() {
+ return jobInfo;
}
@Override
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java
index 98718fe3ba8..b92009422ee 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.operators.coordination;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.metrics.groups.OperatorCoordinatorMetricGroup;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
import org.apache.flink.runtime.jobgraph.OperatorID;
@@ -249,6 +250,11 @@ public class RecreateOnResetOperatorCoordinator implements
OperatorCoordinator {
return context.getJobID();
}
+ @Override
+ public JobInfo getJobInfo() {
+ return context.getJobInfo();
+ }
+
@Override
public OperatorID getOperatorId() {
return context.getOperatorId();
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
index 3f9d78b5864..bb2d6a30050 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.connector.source.ReaderInfo;
import org.apache.flink.api.connector.source.SourceEvent;
import org.apache.flink.api.connector.source.SourceSplit;
@@ -181,6 +182,11 @@ public class SourceCoordinatorContext<SplitT extends
SourceSplit>
return new
InternalSplitEnumeratorMetricGroup(operatorCoordinatorContext.metricGroup());
}
+ @Override
+ public JobInfo getJobInfo() {
+ return operatorCoordinatorContext.getJobInfo();
+ }
+
@Override
public void sendEventToSourceReader(int subtaskId, SourceEvent event) {
checkAndLazyInitialize();
diff --git
a/flink-runtime/src/test/java/org/apache/flink/api/connector/source/RichSourceReaderContextTest.java
b/flink-runtime/src/test/java/org/apache/flink/api/connector/source/RichSourceReaderContextTest.java
new file mode 100644
index 00000000000..79a9d125c14
--- /dev/null
+++
b/flink-runtime/src/test/java/org/apache/flink/api/connector/source/RichSourceReaderContextTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.api.connector.source;
+
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.groups.SourceReaderMetricGroup;
+import org.apache.flink.streaming.util.MockStreamingRuntimeContext;
+import org.apache.flink.util.UserCodeClassLoader;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Unit test for {@link RichSourceReaderContext}. */
+class RichSourceReaderContextTest {
+
+ @Test
+ void testGetJobInfoReturnsRuntimeContextJobInfo() {
+ MockStreamingRuntimeContext runtimeContext = new
MockStreamingRuntimeContext(1, 0);
+ StubRichSourceReaderContext context = new
StubRichSourceReaderContext();
+ context.setRuntimeContext(runtimeContext);
+
+ JobInfo jobInfo = context.getJobInfo();
+
+ assertThat(jobInfo).isSameAs(runtimeContext.getJobInfo());
+ }
+
+ @Test
+ void testGetJobInfoFailsWhenRuntimeContextNotSet() {
+ StubRichSourceReaderContext context = new
StubRichSourceReaderContext();
+ assertThatThrownBy(context::getJobInfo)
+ .hasMessageContaining("runtime context must not be null");
+ }
+
+ private static class StubRichSourceReaderContext extends
RichSourceReaderContext {
+ @Override
+ public SourceReaderMetricGroup metricGroup() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Configuration getConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getLocalHostName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getIndexOfSubtask() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void sendSplitRequest() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void sendSourceEventToCoordinator(SourceEvent sourceEvent) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/MockOperatorCoordinatorContext.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/MockOperatorCoordinatorContext.java
index 71732d8cc7d..6f01ee65629 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/MockOperatorCoordinatorContext.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/MockOperatorCoordinatorContext.java
@@ -19,6 +19,8 @@ limitations under the License.
package org.apache.flink.runtime.operators.coordination;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.JobInfoImpl;
import org.apache.flink.metrics.groups.OperatorCoordinatorMetricGroup;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
import org.apache.flink.runtime.jobgraph.OperatorID;
@@ -31,6 +33,7 @@ import java.util.concurrent.CompletableFuture;
public class MockOperatorCoordinatorContext implements
OperatorCoordinator.Context {
private final JobID jobID = new JobID();
+ private final JobInfo jobInfo = new JobInfoImpl(jobID, "mock-job");
private final OperatorID operatorID;
private final ClassLoader userCodeClassLoader;
private final int numSubtasks;
@@ -62,6 +65,11 @@ public class MockOperatorCoordinatorContext implements
OperatorCoordinator.Conte
return jobID;
}
+ @Override
+ public JobInfo getJobInfo() {
+ return jobInfo;
+ }
+
@Override
public OperatorID getOperatorId() {
return operatorID;
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolderTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolderTest.java
index 8014532c340..d99cc194630 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolderTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolderTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.operators.coordination;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfoImpl;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
@@ -541,6 +542,7 @@ class OperatorCoordinatorHolderTest {
1,
NoOpInvokable.class.getName(),
new Configuration()),
+ new JobInfoImpl(new JobID(), "test-job"),
UnregisteredMetricGroups.createUnregisteredJobManagerJobMetricGroup());
JobID jobId = new JobID();
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
index f593082f6c7..b9ff24d7c58 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
@@ -1194,6 +1194,11 @@ class ExecutingTest {
return null;
}
+ @Override
+ public String getJobName() {
+ return null;
+ }
+
@Override
public BlobWriter getBlobWriter() {
return null;
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
index c4fb85bedce..f6db6e5bb8e 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.source.coordinator;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.connector.source.ReaderInfo;
import org.apache.flink.api.connector.source.SplitsAssignment;
import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
@@ -45,6 +46,16 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Unit test for {@link SourceCoordinatorContext}. */
class SourceCoordinatorContextTest extends SourceCoordinatorTestBase {
+ @Test
+ void testGetJobInfo() {
+ // Compare on JobID (and JobName) rather than full JobInfo because
JobInfoImpl is
+ // @Internal and does not override equals().
+ JobInfo expected = operatorCoordinatorContext.getJobInfo();
+ JobInfo actual = context.getJobInfo();
+ assertThat(actual.getJobId()).isEqualTo(expected.getJobId());
+ assertThat(actual.getJobName()).isEqualTo(expected.getJobName());
+ }
+
@Test
void testRegisterReader() throws Exception {
sourceReady();
diff --git
a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
index 4855ced967b..47a9b54b553 100644
---
a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
+++
b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
@@ -18,6 +18,9 @@
package org.apache.flink.connector.testutils.source.reader;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.JobInfoImpl;
import org.apache.flink.api.connector.source.SourceEvent;
import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.configuration.Configuration;
@@ -36,6 +39,8 @@ public class TestingReaderContext implements
SourceReaderContext {
private final Configuration config;
+ private final JobInfo jobInfo = new JobInfoImpl(new JobID(), "test-job");
+
private final ArrayList<SourceEvent> sentEvents = new ArrayList<>();
private int numSplitRequests;
@@ -51,6 +56,11 @@ public class TestingReaderContext implements
SourceReaderContext {
// ------------------------------------------------------------------------
+ @Override
+ public JobInfo getJobInfo() {
+ return jobInfo;
+ }
+
@Override
public SourceReaderMetricGroup metricGroup() {
return metrics;
diff --git
a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingSplitEnumeratorContext.java
b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingSplitEnumeratorContext.java
index 8b4997f78f1..10e560f72c0 100644
---
a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingSplitEnumeratorContext.java
+++
b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingSplitEnumeratorContext.java
@@ -18,6 +18,9 @@
package org.apache.flink.connector.testutils.source.reader;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.JobInfoImpl;
import org.apache.flink.api.connector.source.ReaderInfo;
import org.apache.flink.api.connector.source.SourceEvent;
import org.apache.flink.api.connector.source.SourceSplit;
@@ -57,6 +60,8 @@ public class TestingSplitEnumeratorContext<SplitT extends
SourceSplit>
private final int parallelism;
+ private final JobInfo jobInfo = new JobInfoImpl(new JobID(), "test-job");
+
public TestingSplitEnumeratorContext(int parallelism) {
this.parallelism = parallelism;
}
@@ -91,6 +96,11 @@ public class TestingSplitEnumeratorContext<SplitT extends
SourceSplit>
// SplitEnumeratorContext methods
// ------------------------------------------------------------------------
+ @Override
+ public JobInfo getJobInfo() {
+ return jobInfo;
+ }
+
@Override
public SplitEnumeratorMetricGroup metricGroup() {
return UnregisteredMetricsGroup.createSplitEnumeratorMetricGroup();
