You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ji...@apache.org on 2023/12/18 13:54:06 UTC
(flink) branch master updated: [FLINK-32949][streaming] Add new taskmanager.collect-sink.port TaskManagerOptions. CollectSinkFunction$ServerThread uses the port configured in TaskManagerOptions#COLLECT_PORT.
This is an automated email from the ASF dual-hosted git repository.
jingge 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 e6169cee12d [FLINK-32949][streaming] Add new taskmanager.collect-sink.port TaskManagerOptions. CollectSinkFunction$ServerThread uses the port configured in TaskManagerOptions#COLLECT_PORT.
e6169cee12d is described below
commit e6169cee12d66ce22f7907b8dc947d4e96228b30
Author: jingge <ge...@gmail.com>
AuthorDate: Tue Dec 5 00:03:38 2023 +0100
[FLINK-32949][streaming]
Add new taskmanager.collect-sink.port TaskManagerOptions.
CollectSinkFunction$ServerThread uses the port configured in TaskManagerOptions#COLLECT_PORT.
---
.../generated/all_taskmanager_section.html | 6 +++
.../generated/common_host_port_section.html | 6 +++
.../generated/task_manager_configuration.html | 6 +++
.../flink/configuration/TaskManagerOptions.java | 13 ++++++
.../api/operators/collect/CollectSinkFunction.java | 31 +++++++++-----
.../operators/collect/CollectSinkFunctionTest.java | 50 ++++++++++++++++------
.../utils/CollectSinkFunctionTestWrapper.java | 5 +++
7 files changed, 92 insertions(+), 25 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/all_taskmanager_section.html b/docs/layouts/shortcodes/generated/all_taskmanager_section.html
index 1f60107c70b..0e7ba018327 100644
--- a/docs/layouts/shortcodes/generated/all_taskmanager_section.html
+++ b/docs/layouts/shortcodes/generated/all_taskmanager_section.html
@@ -32,6 +32,12 @@
<td>String</td>
<td>The local address of the network interface that the task manager binds to. If not configured, '0.0.0.0' will be used.</td>
</tr>
+ <tr>
+ <td><h5>taskmanager.collect-sink.port</h5></td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Integer</td>
+ <td>The port used for the client to retrieve query results from the TaskManager. The default value is 0, which corresponds to a random port assignment.</td>
+ </tr>
<tr>
<td><h5>taskmanager.data.bind-port</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/docs/layouts/shortcodes/generated/common_host_port_section.html b/docs/layouts/shortcodes/generated/common_host_port_section.html
index 160cdd5328d..3c2c318fe6c 100644
--- a/docs/layouts/shortcodes/generated/common_host_port_section.html
+++ b/docs/layouts/shortcodes/generated/common_host_port_section.html
@@ -74,6 +74,12 @@
<td>String</td>
<td>The local address of the network interface that the task manager binds to. If not configured, '0.0.0.0' will be used.</td>
</tr>
+ <tr>
+ <td><h5>taskmanager.collect-sink.port</h5></td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Integer</td>
+ <td>The port used for the client to retrieve query results from the TaskManager. The default value is 0, which corresponds to a random port assignment.</td>
+ </tr>
<tr>
<td><h5>taskmanager.data.bind-port</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/docs/layouts/shortcodes/generated/task_manager_configuration.html b/docs/layouts/shortcodes/generated/task_manager_configuration.html
index 6f7ecf554e0..d9ba88f35c0 100644
--- a/docs/layouts/shortcodes/generated/task_manager_configuration.html
+++ b/docs/layouts/shortcodes/generated/task_manager_configuration.html
@@ -32,6 +32,12 @@
<td>String</td>
<td>The local address of the network interface that the task manager binds to. If not configured, '0.0.0.0' will be used.</td>
</tr>
+ <tr>
+ <td><h5>taskmanager.collect-sink.port</h5></td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Integer</td>
+ <td>The port used for the client to retrieve query results from the TaskManager. The default value is 0, which corresponds to a random port assignment.</td>
+ </tr>
<tr>
<td><h5>taskmanager.debug.memory.log</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
index 5c8d33d7040..2c0a9801ac4 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
@@ -173,6 +173,19 @@ public class TaskManagerOptions {
+ RPC_PORT.key()
+ "') will be used.");
+ /** The default port that <code>CollectSinkFunction$ServerThread</code> is using. */
+ @Documentation.Section({
+ Documentation.Sections.COMMON_HOST_PORT,
+ Documentation.Sections.ALL_TASK_MANAGER
+ })
+ public static final ConfigOption<Integer> COLLECT_PORT =
+ key("taskmanager.collect-sink.port")
+ .intType()
+ .defaultValue(0)
+ .withDescription(
+ "The port used for the client to retrieve query results from the TaskManager. "
+ + "The default value is 0, which corresponds to a random port assignment.");
+
/**
* The initial registration backoff between two consecutive registration attempts. The backoff
* is doubled for each new registration attempt until it reaches the maximum registration
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunction.java
index aeecb135e88..139a7ebf787 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunction.java
@@ -28,6 +28,7 @@ import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
@@ -371,7 +372,7 @@ public class CollectSinkFunction<IN> extends RichSinkFunction<IN>
private ServerThread(TypeSerializer<IN> serializer) throws Exception {
this.serializer = serializer.duplicate();
- this.serverSocket = new ServerSocket(0, 0, getBindAddress());
+ this.serverSocket = new ServerSocket(getPort(), 0, getBindAddress());
this.running = true;
}
@@ -468,22 +469,14 @@ public class CollectSinkFunction<IN> extends RichSinkFunction<IN>
}
private InetSocketAddress getServerSocketAddress() {
- RuntimeContext context = getRuntimeContext();
- Preconditions.checkState(
- context instanceof StreamingRuntimeContext,
- "CollectSinkFunction can only be used in StreamTask");
- StreamingRuntimeContext streamingContext = (StreamingRuntimeContext) context;
+ StreamingRuntimeContext streamingContext = getStreamingRuntimeContext();
String taskManagerAddress =
streamingContext.getTaskManagerRuntimeInfo().getTaskManagerExternalAddress();
return new InetSocketAddress(taskManagerAddress, serverSocket.getLocalPort());
}
private InetAddress getBindAddress() {
- RuntimeContext context = getRuntimeContext();
- Preconditions.checkState(
- context instanceof StreamingRuntimeContext,
- "CollectSinkFunction can only be used in StreamTask");
- StreamingRuntimeContext streamingContext = (StreamingRuntimeContext) context;
+ StreamingRuntimeContext streamingContext = getStreamingRuntimeContext();
String bindAddress =
streamingContext.getTaskManagerRuntimeInfo().getTaskManagerBindAddress();
@@ -497,6 +490,22 @@ public class CollectSinkFunction<IN> extends RichSinkFunction<IN>
return null;
}
+ private int getPort() {
+ return getStreamingRuntimeContext()
+ .getTaskManagerRuntimeInfo()
+ .getConfiguration()
+ .getInteger(TaskManagerOptions.COLLECT_PORT);
+ }
+
+ private StreamingRuntimeContext getStreamingRuntimeContext() {
+ RuntimeContext context = getRuntimeContext();
+ Preconditions.checkState(
+ context instanceof StreamingRuntimeContext,
+ "CollectSinkFunction can only be used in StreamTask");
+ StreamingRuntimeContext streamingContext = (StreamingRuntimeContext) context;
+ return streamingContext;
+ }
+
private void sendBackResults(List<byte[]> serializedResults) throws IOException {
if (LOG.isDebugEnabled()) {
LOG.debug("Sending back " + serializedResults.size() + " results");
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunctionTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunctionTest.java
index d8097b48e2d..d99db8e2057 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunctionTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunctionTest.java
@@ -19,19 +19,25 @@ package org.apache.flink.streaming.api.operators.collect;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.core.testutils.FlinkAssertions;
import org.apache.flink.streaming.api.operators.collect.utils.CollectSinkFunctionTestWrapper;
import org.apache.flink.streaming.api.operators.collect.utils.CollectTestUtils;
import org.apache.flink.util.TestLogger;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.net.BindException;
+import java.net.ServerSocket;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
/** Tests for {@link CollectSinkFunction}. */
public class CollectSinkFunctionTest extends TestLogger {
@@ -39,19 +45,19 @@ public class CollectSinkFunctionTest extends TestLogger {
private CollectSinkFunctionTestWrapper<Integer> functionWrapper;
- @Before
+ @BeforeEach
public void before() throws Exception {
// max bytes per batch = 3 * sizeof(int)
functionWrapper = new CollectSinkFunctionTestWrapper<>(serializer, 12);
}
- @After
+ @AfterEach
public void after() throws Exception {
functionWrapper.closeWrapper();
}
@Test
- public void testIncreasingToken() throws Exception {
+ void testIncreasingToken() throws Exception {
functionWrapper.openFunction();
for (int i = 0; i < 6; i++) {
functionWrapper.invoke(i);
@@ -70,7 +76,7 @@ public class CollectSinkFunctionTest extends TestLogger {
}
@Test
- public void testDuplicatedToken() throws Exception {
+ void testDuplicatedToken() throws Exception {
functionWrapper.openFunction();
for (int i = 0; i < 6; i++) {
functionWrapper.invoke(i);
@@ -89,7 +95,7 @@ public class CollectSinkFunctionTest extends TestLogger {
}
@Test
- public void testInvalidToken() throws Exception {
+ void testInvalidToken() throws Exception {
functionWrapper.openFunction();
for (int i = 0; i < 6; i++) {
functionWrapper.invoke(i);
@@ -106,7 +112,7 @@ public class CollectSinkFunctionTest extends TestLogger {
}
@Test
- public void testInvalidVersion() throws Exception {
+ void testInvalidVersion() throws Exception {
functionWrapper.openFunction();
for (int i = 0; i < 6; i++) {
functionWrapper.invoke(i);
@@ -122,7 +128,23 @@ public class CollectSinkFunctionTest extends TestLogger {
}
@Test
- public void testCheckpoint() throws Exception {
+ void testConfiguredPortIsUsed() throws Exception {
+ int bindingPort = 50500;
+ functionWrapper
+ .getRuntimeContext()
+ .getTaskManagerRuntimeInfo()
+ .getConfiguration()
+ .setInteger(TaskManagerOptions.COLLECT_PORT, bindingPort);
+ functionWrapper.openFunction();
+ assertThatThrownBy(() -> new ServerSocket(bindingPort))
+ .satisfies(
+ FlinkAssertions.anyCauseMatches(
+ BindException.class, "Address already in use (Bind failed)"));
+ functionWrapper.closeFunctionNormally();
+ }
+
+ @Test
+ void testCheckpoint() throws Exception {
functionWrapper.openFunctionWithState();
for (int i = 0; i < 2; i++) {
functionWrapper.invoke(i);
@@ -156,7 +178,7 @@ public class CollectSinkFunctionTest extends TestLogger {
}
@Test
- public void testRestart() throws Exception {
+ void testRestart() throws Exception {
functionWrapper.openFunctionWithState();
for (int i = 0; i < 3; i++) {
functionWrapper.invoke(i);
@@ -195,12 +217,12 @@ public class CollectSinkFunctionTest extends TestLogger {
}
@Test
- public void testAccumulatorResultWithoutCheckpoint() throws Exception {
+ void testAccumulatorResultWithoutCheckpoint() throws Exception {
testAccumulatorResultWithoutCheckpoint(2, Arrays.asList(2, 3, 4, 5));
}
@Test
- public void testEmptyAccumulatorResult() throws Exception {
+ void testEmptyAccumulatorResult() throws Exception {
testAccumulatorResultWithoutCheckpoint(6, Collections.emptyList());
}
@@ -219,7 +241,7 @@ public class CollectSinkFunctionTest extends TestLogger {
}
@Test
- public void testAccumulatorResultWithCheckpoint() throws Exception {
+ void testAccumulatorResultWithCheckpoint() throws Exception {
functionWrapper.openFunctionWithState();
for (int i = 0; i < 6; i++) {
functionWrapper.invoke(i);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/utils/CollectSinkFunctionTestWrapper.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/utils/CollectSinkFunctionTestWrapper.java
index 5e193ee70bc..4ac3a2e0ff3 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/utils/CollectSinkFunctionTestWrapper.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/utils/CollectSinkFunctionTestWrapper.java
@@ -58,6 +58,7 @@ public class CollectSinkFunctionTestWrapper<IN> {
private final int maxBytesPerBatch;
private final IOManager ioManager;
+
private final StreamingRuntimeContext runtimeContext;
private final MockOperatorEventGateway gateway;
private final CollectSinkOperatorCoordinator coordinator;
@@ -175,4 +176,8 @@ public class CollectSinkFunctionTestWrapper<IN> {
Accumulator accumulator = runtimeContext.getAccumulator(ACCUMULATOR_NAME);
return ((SerializedListAccumulator) accumulator).getLocalValue();
}
+
+ public StreamingRuntimeContext getRuntimeContext() {
+ return runtimeContext;
+ }
}