Skip to content

[Test] Eliminate OPC UA sink port allocation race #18817

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 1 commit into from
Oct 10, 2026
Merged

[Test] Eliminate OPC UA sink port allocation race #18817

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 18 additions & 61 deletions ...pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,8 +37,6 @@
import org.junit.Test;

import java.io.File;
import java.io.IOException;
import java.net.ServerSocket;
import java.security.SecureRandom;
import java.util.Arrays;
import java.util.HashMap;
Expand Down Expand Up @@ -67,34 +65,15 @@ public void testValidateClientServerOnlyOptions() {

@Test
public void testCustomizeQualityParameters() throws Exception {
final int[] qualityPorts = findTwoFreePorts();
final int[] normalPorts = findTwoFreePorts();
final PipeParameters qualityParameters = createQualityServerParameters(0);
qualityParameters
.getAttribute()
.put(PipeSinkConstant.CONNECTOR_OPC_UA_DEFAULT_QUALITY_KEY, "BAD");
try (final OpcUaSink qualitySink = new OpcUaSink();
final OpcUaSink normalSink = new OpcUaSink()) {
qualitySink.customize(
createParameters(
PipeSinkConstant.CONNECTOR_KEY,
PipeSinkConstant.OPC_UA_SINK_NAME,
PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_KEY,
"true",
PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_KEY,
"value1",
PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_NAME_KEY,
"quality1",
PipeSinkConstant.CONNECTOR_OPC_UA_DEFAULT_QUALITY_KEY,
"BAD",
PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_POLICY_KEY,
"None",
PipeSinkConstant.CONNECTOR_OPC_UA_TCP_BIND_PORT_KEY,
Integer.toString(qualityPorts[0]),
PipeSinkConstant.CONNECTOR_OPC_UA_HTTPS_BIND_PORT_KEY,
Integer.toString(qualityPorts[1]),
PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_DIR_KEY,
createSecurityDir()),
createRuntimeConfiguration());
qualitySink.customize(qualityParameters, createRuntimeConfiguration());
normalSink.customize(
createOpcUaServerParameters(normalPorts[0], normalPorts[1], "root", "root"),
createRuntimeConfiguration());
createOpcUaServerParameters(1, "root", "root"), createRuntimeConfiguration());

Assert.assertEquals("value1", qualitySink.getValueName());
Assert.assertEquals("quality1", qualitySink.getQualityName());
Expand Down Expand Up @@ -136,16 +115,11 @@ public void testTransferWithQualityAndNormalTablets() throws Exception {
qualityTablet.addValue(opcSchemaList.get(1).getMeasurementName(), rowSize, true);
}

final int[] qualityPorts = findTwoFreePorts();
final int[] normalPorts = findTwoFreePorts();
try (final OpcUaSink qualitySink = new OpcUaSink();
final OpcUaSink normalSink = new OpcUaSink()) {
qualitySink.customize(
createQualityServerParameters(qualityPorts[0], qualityPorts[1]),
createRuntimeConfiguration());
qualitySink.customize(createQualityServerParameters(0), createRuntimeConfiguration());
normalSink.customize(
createOpcUaServerParameters(normalPorts[0], normalPorts[1], "root", "root"),
createRuntimeConfiguration());
createOpcUaServerParameters(1, "root", "root"), createRuntimeConfiguration());

final PipeRawTabletInsertionEvent event =
new PipeRawTabletInsertionEvent(
Expand Down Expand Up @@ -173,12 +147,9 @@ public void testTransferWithQualityAndNormalTablets() throws Exception {

@Test
public void testSharedServerLifecycle() throws Exception {
final int[] ports = findTwoFreePorts();
final PipeTaskRuntimeConfiguration configuration = createRuntimeConfiguration();
final PipeParameters parameters =
createOpcUaServerParameters(ports[0], ports[1], "root", "root");
final PipeParameters conflictingParameters =
createOpcUaServerParameters(ports[0], ports[1], "root", "conflict");
final PipeParameters parameters = createOpcUaServerParameters(0, "root", "root");
final PipeParameters conflictingParameters = createOpcUaServerParameters(0, "root", "conflict");

try (final OpcUaSink firstSink = new OpcUaSink();
final OpcUaSink secondSink = new OpcUaSink()) {
Expand Down Expand Up @@ -214,26 +185,27 @@ private static void assertValidationFailure(
Assert.assertTrue(exception.getMessage(), exception.getMessage().contains(expectedMessagePart));
}

private static PipeParameters createQualityServerParameters(
final int tcpPort, final int httpsPort) {
private static PipeParameters createQualityServerParameters(final int serverId) {
final Map<String, String> attributes =
new HashMap<>(
createOpcUaServerParameters(tcpPort, httpsPort, "root", "root").getAttribute());
new HashMap<>(createOpcUaServerParameters(serverId, "root", "root").getAttribute());
attributes.put(PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_KEY, "true");
attributes.put(PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_KEY, "value1");
attributes.put(PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_NAME_KEY, "quality1");
return new PipeParameters(attributes);
}

private static PipeParameters createOpcUaServerParameters(
final int tcpPort, final int httpsPort, final String user, final String password) {
final int serverId, final String user, final String password) {
// Bind port 0 directly so the OS allocates the TCP port atomically. HTTPS is not bound by the
// TCP-only transport, but its configured port still distinguishes the sink's shared server
// keys.
return createParameters(
PipeSinkConstant.CONNECTOR_KEY,
PipeSinkConstant.OPC_UA_SINK_NAME,
PipeSinkConstant.CONNECTOR_OPC_UA_TCP_BIND_PORT_KEY,
Integer.toString(tcpPort),
"0",
PipeSinkConstant.CONNECTOR_OPC_UA_HTTPS_BIND_PORT_KEY,
Integer.toString(httpsPort),
Integer.toString(serverId),
PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_POLICY_KEY,
"None",
PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY,
Expand Down Expand Up @@ -267,19 +239,4 @@ private static String createSecurityDir() {
+ UUID.randomUUID())
.getAbsolutePath();
}

private static int[] findTwoFreePorts() throws IOException {
final int firstPort = findFreePort();
int secondPort;
do {
secondPort = findFreePort();
} while (secondPort == firstPort);
return new int[] {firstPort, secondPort};
}

private static int findFreePort() throws IOException {
try (final ServerSocket socket = new ServerSocket(0)) {
return socket.getLocalPort();
}
}
}
Loading