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/19 21:11:43 UTC
(flink) 01/02: [FLINK-33877][streaming] Fix unstable test of CollectSinkFunctionTest.testConfiguredPortIsUsed
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
commit 409f44660eb434a57237af2690c9e4b7f9679442
Author: Jiabao Sun <ji...@xtransfer.cn>
AuthorDate: Tue Dec 19 11:40:51 2023 +0800
[FLINK-33877][streaming] Fix unstable test of CollectSinkFunctionTest.testConfiguredPortIsUsed
---
.../operators/collect/CollectSinkFunctionTest.java | 25 +++++++++++-----------
1 file changed, 13 insertions(+), 12 deletions(-)
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 d99db8e2057..dd22199e0a8 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
@@ -129,18 +129,19 @@ public class CollectSinkFunctionTest extends TestLogger {
@Test
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();
+ try (ServerSocket socket = new ServerSocket(0)) {
+ functionWrapper
+ .getRuntimeContext()
+ .getTaskManagerRuntimeInfo()
+ .getConfiguration()
+ .setInteger(TaskManagerOptions.COLLECT_PORT, socket.getLocalPort());
+ assertThatThrownBy(() -> functionWrapper.openFunction())
+ .satisfies(
+ FlinkAssertions.anyCauseMatches(
+ BindException.class, "Address already in use (Bind failed)"));
+ } finally {
+ functionWrapper.closeWrapper();
+ }
}
@Test