You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by jq...@apache.org on 2020/10/28 01:10:33 UTC
[flink] branch master updated: [FLINK-18044][Connectors/Common] Add
the subtask index information to the SourceReaderContext. (#12647)
This is an automated email from the ASF dual-hosted git repository.
jqin 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 fc41f03 [FLINK-18044][Connectors/Common] Add the subtask index information to the SourceReaderContext. (#12647)
fc41f03 is described below
commit fc41f033a0990afe8bdec272c0350705a9e23146
Author: fangliang <56...@qq.com>
AuthorDate: Wed Oct 28 09:08:38 2020 +0800
[FLINK-18044][Connectors/Common] Add the subtask index information to the SourceReaderContext. (#12647)
---
.../flink/connector/file/src/FileSourceHeavyThroughputTest.java | 5 +++++
.../org/apache/flink/api/connector/source/SourceReaderContext.java | 5 +++++
.../flink/api/connector/source/mocks/TestingReaderContext.java | 5 +++++
.../org/apache/flink/streaming/api/operators/SourceOperator.java | 5 +++++
4 files changed, 20 insertions(+)
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/FileSourceHeavyThroughputTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/FileSourceHeavyThroughputTest.java
index ec62f97..b54c7f3 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/FileSourceHeavyThroughputTest.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/FileSourceHeavyThroughputTest.java
@@ -202,6 +202,11 @@ public class FileSourceHeavyThroughputTest {
}
@Override
+ public int getIndexOfSubtask() {
+ return 0;
+ }
+
+ @Override
public void sendSourceEventToCoordinator(SourceEvent sourceEvent) {}
}
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 c8ed36d..4d9e53a 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
@@ -45,6 +45,11 @@ public interface SourceReaderContext {
String getLocalHostName();
/**
+ * @return The index of this subtask.
+ */
+ int getIndexOfSubtask();
+
+ /**
* Send a source event to the source coordinator.
*
* @param sourceEvent the source event to coordinator.
diff --git a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/TestingReaderContext.java b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/TestingReaderContext.java
index cee4654..3f254e4 100644
--- a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/TestingReaderContext.java
+++ b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/TestingReaderContext.java
@@ -64,6 +64,11 @@ public class TestingReaderContext implements SourceReaderContext {
}
@Override
+ public int getIndexOfSubtask() {
+ return 0;
+ }
+
+ @Override
public void sendSourceEventToCoordinator(SourceEvent sourceEvent) {
sentEvents.add(sourceEvent);
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
index 95c9eda..742d7ce 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
@@ -150,6 +150,11 @@ public class SourceOperator<OUT, SplitT extends SourceSplit>
}
@Override
+ public int getIndexOfSubtask() {
+ return getRuntimeContext().getIndexOfThisSubtask();
+ }
+
+ @Override
public void sendSourceEventToCoordinator(SourceEvent event) {
operatorEventGateway.sendEventToCoordinator(new SourceEventWrapper(event));
}