This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 3048642177 [Improve][Connector-V2] Migrate EdgeSocket validation to
declarative OptionRule (#11384)
3048642177 is described below
commit 304864217759e227fbbf384329eee0a046c2a33f
Author: zhiwei.niu <[email protected]>
AuthorDate: Sun Jul 12 22:47:46 2026 +0800
[Improve][Connector-V2] Migrate EdgeSocket validation to declarative
OptionRule (#11384)
---
docs/en/connectors/source/EdgeSocket.md | 1 +
docs/zh/connectors/source/EdgeSocket.md | 1 +
.../edgesocket/config/EdgeSocketConfig.java | 50 +--------
.../edgesocket/config/EdgeSocketSourceOptions.java | 26 +++--
.../edgesocket/source/EdgeSocketSourceFactory.java | 109 +++++++++++++++++-
.../edgesocket/EdgeSocketFactoryTest.java | 124 ++++++++++++++++++++-
.../edgesocket/config/EdgeSocketConfigTest.java | 118 --------------------
.../connector/edgesocket/AbstractEdgeSocketIT.java | 78 +++++++++++++
.../connector/edgesocket/EdgeSocketMysql8_4IT.java | 37 ++++++
9 files changed, 356 insertions(+), 188 deletions(-)
diff --git a/docs/en/connectors/source/EdgeSocket.md
b/docs/en/connectors/source/EdgeSocket.md
index ad701d5759..b56c5d469a 100644
--- a/docs/en/connectors/source/EdgeSocket.md
+++ b/docs/en/connectors/source/EdgeSocket.md
@@ -360,6 +360,7 @@ The following table lists connection outcomes and every
application-level respon
The diagram below shows the full collector lifecycle with all response
branches and corresponding actions.
```mermaid
+%%{init: {"theme": "base", "themeVariables": {"background": "#0f1d33",
"primaryColor": "#0c2530", "primaryBorderColor": "#2dd4bf", "primaryTextColor":
"#f8fbff", "actorBkg": "#0c2530", "actorBorder": "#2dd4bf", "actorTextColor":
"#f8fbff", "activationBkgColor": "#1f1a34", "activationBorderColor": "#8d7cf6",
"noteBkgColor": "#1f1a34", "noteBorderColor": "#8d7cf6", "noteTextColor":
"#f8fbff", "signalColor": "#5db8e2", "signalTextColor": "#f8fbff",
"labelBoxBkgColor": "#0f1d33", "labelBoxBo [...]
sequenceDiagram
participant C as Collector
participant S as EdgeSocket Source
diff --git a/docs/zh/connectors/source/EdgeSocket.md
b/docs/zh/connectors/source/EdgeSocket.md
index 69a61b0d4a..51bea57f3e 100644
--- a/docs/zh/connectors/source/EdgeSocket.md
+++ b/docs/zh/connectors/source/EdgeSocket.md
@@ -357,6 +357,7 @@ Source 回复：
下图展示采集端完整生命周期，包含所有响应分支和对应操作。
```mermaid
+%%{init: {"theme": "base", "themeVariables": {"background": "#0f1d33",
"primaryColor": "#0c2530", "primaryBorderColor": "#2dd4bf", "primaryTextColor":
"#f8fbff", "actorBkg": "#0c2530", "actorBorder": "#2dd4bf", "actorTextColor":
"#f8fbff", "activationBkgColor": "#1f1a34", "activationBorderColor": "#8d7cf6",
"noteBkgColor": "#1f1a34", "noteBorderColor": "#8d7cf6", "noteTextColor":
"#f8fbff", "signalColor": "#5db8e2", "signalTextColor": "#f8fbff",
"labelBoxBkgColor": "#0f1d33", "labelBoxBo [...]
sequenceDiagram
participant C as Collector
participant S as EdgeSocket Source
diff --git
a/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketConfig.java
b/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketConfig.java
index 048be6d25e..b51357923e 100644
---
a/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketConfig.java
+++
b/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketConfig.java
@@ -52,61 +52,15 @@ public class EdgeSocketConfig implements Serializable {
this.maxNumRetries = config.get(EdgeSocketSourceOptions.MAX_RETRIES);
this.reconnectIntervalMs =
config.get(EdgeSocketSourceOptions.RECONNECT_INTERVAL_MS);
this.acceptTimeoutMs =
config.get(EdgeSocketSourceOptions.ACCEPT_TIMEOUT_MS);
- this.packetMode =
-
EdgeSocketPacketMode.from(config.get(EdgeSocketSourceOptions.PACKET_MODE));
+ this.packetMode = config.get(EdgeSocketSourceOptions.PACKET_MODE);
this.secretKey =
config.getOptional(EdgeSocketSourceOptions.SECRET_KEY).orElse(null);
- this.authType =
EdgeSocketAuthType.from(config.get(EdgeSocketSourceOptions.AUTH_TYPE));
+ this.authType = config.get(EdgeSocketSourceOptions.AUTH_TYPE);
this.token =
config.getOptional(EdgeSocketSourceOptions.TOKEN).orElse(null);
if (this.endpoint != null && this.endpoint.trim().isEmpty()) {
this.endpoint = null;
}
- if (this.endpoint != null) {
- int separatorIndex = this.endpoint.lastIndexOf(':');
- if (separatorIndex <= 0 || separatorIndex >=
this.endpoint.length() - 1) {
- throw new IllegalArgumentException(
- "Invalid endpoint: " + this.endpoint + ", expected
format host:port");
- }
- String endpointPort = this.endpoint.substring(separatorIndex + 1);
- try {
- Integer.parseInt(endpointPort);
- } catch (NumberFormatException parseException) {
- throw new IllegalArgumentException(
- "Invalid endpoint port in endpoint: " + this.endpoint,
parseException);
- }
- }
- if (this.localQueueCapacity <= 0) {
- throw new IllegalArgumentException(
- String.format(
- "Invalid local_queue_capacity: %s, it must be
greater than 0",
- this.localQueueCapacity));
- }
- if (this.queueBackpressureWatermarkRatio <= 0 ||
this.queueBackpressureWatermarkRatio > 1) {
- throw new IllegalArgumentException(
- String.format(
- "Invalid queue_backpressure_watermark_ratio: %s, "
- + "it must be in (0, 1]",
- this.queueBackpressureWatermarkRatio));
- }
- if (this.queueFullRetryAfterMs <= 0) {
- throw new IllegalArgumentException(
- String.format(
- "Invalid queue_full_retry_after_ms: %s, it must be
greater than 0",
- this.queueFullRetryAfterMs));
- }
if (this.packetMode == EdgeSocketPacketMode.PACKET && this.secretKey
!= null) {
this.secretKeyBytes = Base64.getDecoder().decode(this.secretKey);
- if (this.secretKeyBytes.length != 32) {
- throw new IllegalArgumentException(
- String.format(
- "Invalid secret_key: AES-256 requires exactly
32 bytes, "
- + "but got %d bytes after Base64
decoding",
- this.secretKeyBytes.length));
- }
- }
- if (this.authType == EdgeSocketAuthType.TOKEN
- && (this.token == null || this.token.trim().isEmpty())) {
- throw new IllegalArgumentException(
- "Invalid token: token is required and cannot be empty when
auth_type is TOKEN");
}
}
}
diff --git
a/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketSourceOptions.java
b/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketSourceOptions.java
index 2670b36bc6..df726e0acb 100644
---
a/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketSourceOptions.java
@@ -19,6 +19,10 @@ package
org.apache.seatunnel.connectors.seatunnel.edgesocket.config;
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.SingleChoiceOption;
+import
org.apache.seatunnel.connectors.seatunnel.edgesocket.serialize.EdgeSocketPacketMode;
+
+import java.util.Arrays;
public class EdgeSocketSourceOptions extends EdgeSocketCommonOptions {
private static final int DEFAULT_LOCAL_QUEUE_CAPACITY = 1024;
@@ -27,9 +31,6 @@ public class EdgeSocketSourceOptions extends
EdgeSocketCommonOptions {
private static final int DEFAULT_MAX_RETRIES = 3;
private static final int DEFAULT_RECONNECT_INTERVAL_MS = 1000;
private static final int DEFAULT_ACCEPT_TIMEOUT_MS = 1000;
- private static final String DEFAULT_PACKET_MODE = "RAW";
- private static final String DEFAULT_AUTH_TYPE = "TOKEN";
-
public static final Option<Integer> LOCAL_QUEUE_CAPACITY =
Options.key("local_queue_capacity")
.intType()
@@ -80,13 +81,15 @@ public class EdgeSocketSourceOptions extends
EdgeSocketCommonOptions {
"Socket accept timeout in milliseconds, default is
"
+ DEFAULT_ACCEPT_TIMEOUT_MS);
- public static final Option<String> PACKET_MODE =
+ public static final SingleChoiceOption<EdgeSocketPacketMode> PACKET_MODE =
Options.key("packet_mode")
- .stringType()
- .defaultValue(DEFAULT_PACKET_MODE)
+ .singleChoice(
+ EdgeSocketPacketMode.class,
+ Arrays.asList(EdgeSocketPacketMode.values()))
+ .defaultValue(EdgeSocketPacketMode.RAW)
.withDescription(
"Incoming packet mode, supported values: RAW,
PACKET. Default is "
- + DEFAULT_PACKET_MODE);
+ + EdgeSocketPacketMode.RAW);
public static final Option<String> SECRET_KEY =
Options.key("secret_key")
@@ -102,11 +105,12 @@ public class EdgeSocketSourceOptions extends
EdgeSocketCommonOptions {
.withDescription(
"Token value used by TOKEN auth_type. This option
is required when auth_type is TOKEN.");
- public static final Option<String> AUTH_TYPE =
+ public static final SingleChoiceOption<EdgeSocketAuthType> AUTH_TYPE =
Options.key("auth_type")
- .stringType()
- .defaultValue(DEFAULT_AUTH_TYPE)
+ .singleChoice(
+ EdgeSocketAuthType.class,
Arrays.asList(EdgeSocketAuthType.values()))
+ .defaultValue(EdgeSocketAuthType.TOKEN)
.withDescription(
"Authentication type for ingress connection.
Currently supported value: TOKEN. Default is "
- + DEFAULT_AUTH_TYPE);
+ + EdgeSocketAuthType.TOKEN);
}
diff --git
a/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/source/EdgeSocketSourceFactory.java
b/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/source/EdgeSocketSourceFactory.java
index 58fe5c45d3..95f1e09b18 100644
---
a/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/source/EdgeSocketSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/source/EdgeSocketSourceFactory.java
@@ -17,7 +17,11 @@
package org.apache.seatunnel.connectors.seatunnel.edgesocket.source;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.source.SourceSplit;
@@ -26,10 +30,12 @@ import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.api.table.factory.TableSourceFactory;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import
org.apache.seatunnel.connectors.seatunnel.edgesocket.config.EdgeSocketSourceOptions;
+import
org.apache.seatunnel.connectors.seatunnel.edgesocket.serialize.EdgeSocketPacketMode;
import com.google.auto.service.AutoService;
import java.io.Serializable;
+import java.util.Base64;
@AutoService(Factory.class)
public class EdgeSocketSourceFactory implements TableSourceFactory {
@@ -42,19 +48,44 @@ public class EdgeSocketSourceFactory implements
TableSourceFactory {
@Override
public OptionRule optionRule() {
return OptionRule.builder()
- .required(EdgeSocketSourceOptions.PORT,
EdgeSocketSourceOptions.TOKEN)
+ .required(
+ EdgeSocketSourceOptions.PORT,
+ Conditions.greaterThan(EdgeSocketSourceOptions.PORT,
0))
+ .required(
+ EdgeSocketSourceOptions.TOKEN,
+ Conditions.notBlank(EdgeSocketSourceOptions.TOKEN))
.optional(
EdgeSocketSourceOptions.ENDPOINT,
- ConnectorCommonOptions.SCHEMA,
+ Conditions.extension(
+ EdgeSocketSourceOptions.ENDPOINT, new
EndpointValidator()))
+ .optional(ConnectorCommonOptions.SCHEMA)
+ .optional(
EdgeSocketSourceOptions.LOCAL_QUEUE_CAPACITY,
+
Conditions.greaterThan(EdgeSocketSourceOptions.LOCAL_QUEUE_CAPACITY, 0))
+ .optional(
EdgeSocketSourceOptions.QUEUE_BACKPRESSURE_WATERMARK_RATIO,
+ Conditions.greaterThan(
+
EdgeSocketSourceOptions.QUEUE_BACKPRESSURE_WATERMARK_RATIO,
+ 0.0)
+ .and(
+ Conditions.lessOrEqual(
+ EdgeSocketSourceOptions
+
.QUEUE_BACKPRESSURE_WATERMARK_RATIO,
+ 1.0)))
+ .optional(
EdgeSocketSourceOptions.QUEUE_FULL_RETRY_AFTER_MS,
+ Conditions.greaterThan(
+
EdgeSocketSourceOptions.QUEUE_FULL_RETRY_AFTER_MS, 0))
+ .optional(
+ EdgeSocketSourceOptions.SECRET_KEY,
+ Conditions.extension(
+ EdgeSocketSourceOptions.SECRET_KEY, new
SecretKeyValidator()))
+ .optional(
+ EdgeSocketSourceOptions.PACKET_MODE,
+ EdgeSocketSourceOptions.AUTH_TYPE,
EdgeSocketSourceOptions.MAX_RETRIES,
EdgeSocketSourceOptions.RECONNECT_INTERVAL_MS,
- EdgeSocketSourceOptions.ACCEPT_TIMEOUT_MS,
- EdgeSocketSourceOptions.PACKET_MODE,
- EdgeSocketSourceOptions.SECRET_KEY,
- EdgeSocketSourceOptions.AUTH_TYPE)
+ EdgeSocketSourceOptions.ACCEPT_TIMEOUT_MS)
.build();
}
@@ -69,4 +100,70 @@ public class EdgeSocketSourceFactory implements
TableSourceFactory {
public Class<? extends SeaTunnelSource> getSourceClass() {
return EdgeSocketSource.class;
}
+
+ private static class EndpointValidator implements
ConditionExtension<String> {
+
+ @Override
+ public String description() {
+ return "must be blank or in host:port format";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String endpoint) {
+ if (endpoint == null || endpoint.trim().isEmpty()) {
+ return true;
+ }
+ int separatorIndex = endpoint.lastIndexOf(':');
+ if (separatorIndex <= 0 || separatorIndex >= endpoint.length() -
1) {
+ throw new OptionValidationException(
+ "Invalid endpoint: %s, expected format host:port",
endpoint);
+ }
+ String endpointPort = endpoint.substring(separatorIndex + 1);
+ try {
+ Integer.parseInt(endpointPort);
+ } catch (NumberFormatException parseException) {
+ throw new OptionValidationException(
+ String.format("Invalid endpoint port in endpoint: %s",
endpoint),
+ parseException);
+ }
+ return true;
+ }
+ }
+
+ private static class SecretKeyValidator implements
ConditionExtension<String> {
+
+ @Override
+ public String description() {
+ return "must decode to exactly 32 bytes when packet_mode is
PACKET";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String secretKey) {
+ if (secretKey == null) {
+ return true;
+ }
+ EdgeSocketPacketMode packetMode;
+ try {
+ packetMode = config.get(EdgeSocketSourceOptions.PACKET_MODE);
+ } catch (IllegalArgumentException exception) {
+ return true;
+ }
+ if (packetMode != EdgeSocketPacketMode.PACKET) {
+ return true;
+ }
+ byte[] secretKeyBytes;
+ try {
+ secretKeyBytes = Base64.getDecoder().decode(secretKey);
+ } catch (IllegalArgumentException exception) {
+ throw new OptionValidationException("Invalid secret_key: not
Base64 encoded");
+ }
+ if (secretKeyBytes.length != 32) {
+ throw new OptionValidationException(
+ "Invalid secret_key: AES-256 requires exactly 32
bytes, "
+ + "but got %d bytes after Base64 decoding",
+ secretKeyBytes.length);
+ }
+ return true;
+ }
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-edge-socket/src/test/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/EdgeSocketFactoryTest.java
b/seatunnel-connectors-v2/connector-edge-socket/src/test/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/EdgeSocketFactoryTest.java
index 3cc3075242..1985fcab2a 100644
---
a/seatunnel-connectors-v2/connector-edge-socket/src/test/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/EdgeSocketFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-edge-socket/src/test/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/EdgeSocketFactoryTest.java
@@ -18,7 +18,9 @@
package org.apache.seatunnel.connectors.seatunnel.edgesocket;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.Boundedness;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
@@ -59,6 +61,105 @@ class EdgeSocketFactoryTest {
"optionRule must include queue_full_retry_after_ms");
}
+ @Test
+ void optionRuleShouldAcceptValidConfig() {
+ assertValidOptionRule(validConfig());
+ }
+
+ @Test
+ void optionRuleShouldRejectNonPositivePort() {
+ Map<String, Object> configMap = validConfig();
+ configMap.put(EdgeSocketCommonOptions.PORT.key(), 0);
+
+ assertInvalidOptionRule(configMap);
+
+ configMap.put(EdgeSocketCommonOptions.PORT.key(), -1);
+ assertInvalidOptionRule(configMap);
+ }
+
+ @Test
+ void optionRuleShouldRejectBlankToken() {
+ Map<String, Object> configMap = validConfig();
+ configMap.put(EdgeSocketSourceOptions.TOKEN.key(), " ");
+
+ assertInvalidOptionRule(configMap);
+
+ configMap = validConfig();
+ configMap.remove(EdgeSocketSourceOptions.TOKEN.key());
+ assertInvalidOptionRule(configMap);
+ }
+
+ @Test
+ void optionRuleShouldRejectInvalidEndpoint() {
+ Map<String, Object> configMap = validConfig();
+ configMap.put(EdgeSocketCommonOptions.ENDPOINT.key(),
"edge.lb.example.com");
+
+ assertInvalidOptionRule(configMap);
+
+ configMap.put(EdgeSocketCommonOptions.ENDPOINT.key(),
"edge.lb.example.com:abc");
+ assertInvalidOptionRule(configMap);
+ }
+
+ @Test
+ void optionRuleShouldAllowBlankEndpoint() {
+ Map<String, Object> configMap = validConfig();
+ configMap.put(EdgeSocketCommonOptions.ENDPOINT.key(), " ");
+
+ assertValidOptionRule(configMap);
+ }
+
+ @Test
+ void optionRuleShouldRejectInvalidBackpressureOptions() {
+ Map<String, Object> configMap = validConfig();
+ configMap.put(EdgeSocketSourceOptions.LOCAL_QUEUE_CAPACITY.key(), 0);
+ assertInvalidOptionRule(configMap);
+
+ configMap = validConfig();
+
configMap.put(EdgeSocketSourceOptions.QUEUE_BACKPRESSURE_WATERMARK_RATIO.key(),
0.0);
+ assertInvalidOptionRule(configMap);
+
+ configMap = validConfig();
+
configMap.put(EdgeSocketSourceOptions.QUEUE_BACKPRESSURE_WATERMARK_RATIO.key(),
1.01);
+ assertInvalidOptionRule(configMap);
+
+ configMap = validConfig();
+ configMap.put(EdgeSocketSourceOptions.QUEUE_FULL_RETRY_AFTER_MS.key(),
0);
+ assertInvalidOptionRule(configMap);
+ }
+
+ @Test
+ void optionRuleShouldRejectInvalidPacketAndAuthMode() {
+ Map<String, Object> configMap = validConfig();
+ configMap.put(EdgeSocketSourceOptions.PACKET_MODE.key(),
"unsupported");
+ assertInvalidOptionRuleOrType(configMap);
+
+ configMap = validConfig();
+ configMap.put(EdgeSocketSourceOptions.AUTH_TYPE.key(), "unsupported");
+ assertInvalidOptionRuleOrType(configMap);
+ }
+
+ @Test
+ void optionRuleShouldValidateSecretKeyOnlyInPacketMode() {
+ Map<String, Object> configMap = validConfig();
+ configMap.put(EdgeSocketSourceOptions.PACKET_MODE.key(), "PACKET");
+ configMap.put(EdgeSocketSourceOptions.SECRET_KEY.key(), "c2hvcnQ=");
+
+ assertInvalidOptionRule(configMap);
+
+ configMap.put(EdgeSocketSourceOptions.SECRET_KEY.key(),
"invalid-not-base64!!!");
+ assertInvalidOptionRule(configMap);
+
+ configMap.put(
+ EdgeSocketSourceOptions.SECRET_KEY.key(),
+ "dGVzdC1zZWNyZXQta2V5LTMyLWJ5dGVzLWFlczI1NiE=");
+ assertValidOptionRule(configMap);
+
+ configMap = validConfig();
+ configMap.put(EdgeSocketSourceOptions.PACKET_MODE.key(), "RAW");
+ configMap.put(EdgeSocketSourceOptions.SECRET_KEY.key(), "c2hvcnQ=");
+ assertValidOptionRule(configMap);
+ }
+
@Test
void shouldUseNullableEndpointAndQueueCapacity() {
Map<String, Object> configMap = new HashMap<>();
@@ -109,15 +210,28 @@ class EdgeSocketFactoryTest {
Assertions.assertEquals("edge.lb.example.com:10091",
config.getEndpoint());
}
- @Test
- void shouldRejectInvalidEndpoint() {
+ private static Map<String, Object> validConfig() {
Map<String, Object> configMap = new HashMap<>();
configMap.put(EdgeSocketCommonOptions.PORT.key(), 10001);
configMap.put(EdgeSocketSourceOptions.TOKEN.key(), "token");
- configMap.put(EdgeSocketCommonOptions.ENDPOINT.key(),
"edge.lb.example.com");
+ return configMap;
+ }
+ private static void assertValidOptionRule(Map<String, Object> configMap) {
+ Assertions.assertDoesNotThrow(() -> validateOptionRule(configMap));
+ }
+
+ private static void assertInvalidOptionRule(Map<String, Object> configMap)
{
Assertions.assertThrows(
- IllegalArgumentException.class,
- () -> new EdgeSocketConfig(ReadonlyConfig.fromMap(configMap)));
+ OptionValidationException.class, () ->
validateOptionRule(configMap));
+ }
+
+ private static void assertInvalidOptionRuleOrType(Map<String, Object>
configMap) {
+ Assertions.assertThrows(RuntimeException.class, () ->
validateOptionRule(configMap));
+ }
+
+ private static void validateOptionRule(Map<String, Object> configMap) {
+ ConfigValidator.of(ReadonlyConfig.fromMap(configMap))
+ .validate(new EdgeSocketSourceFactory().optionRule());
}
}
diff --git
a/seatunnel-connectors-v2/connector-edge-socket/src/test/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketConfigTest.java
b/seatunnel-connectors-v2/connector-edge-socket/src/test/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketConfigTest.java
index a7404e98e2..1ea1e301da 100644
---
a/seatunnel-connectors-v2/connector-edge-socket/src/test/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketConfigTest.java
+++
b/seatunnel-connectors-v2/connector-edge-socket/src/test/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/config/EdgeSocketConfigTest.java
@@ -46,42 +46,6 @@ class EdgeSocketConfigTest {
Assertions.assertEquals("collector.example.com:8080",
config.getEndpoint());
}
- @Test
- void endpointWithoutPortThrows() {
- Map<String, Object> map = minimalConfig();
- map.put(EdgeSocketCommonOptions.ENDPOINT.key(), "no-port-host");
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
- Assertions.assertTrue(ex.getMessage().contains("expected format
host:port"));
- }
-
- @Test
- void endpointWithNonNumericPortThrows() {
- Map<String, Object> map = minimalConfig();
- map.put(EdgeSocketCommonOptions.ENDPOINT.key(), "host:abc");
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
- Assertions.assertTrue(ex.getMessage().contains("Invalid endpoint
port"));
- }
-
- @Test
- void endpointColonOnlyThrows() {
- Map<String, Object> map = minimalConfig();
- map.put(EdgeSocketCommonOptions.ENDPOINT.key(), ":8080");
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
- Assertions.assertTrue(ex.getMessage().contains("expected format
host:port"));
- }
-
- @Test
- void endpointTrailingColonThrows() {
- Map<String, Object> map = minimalConfig();
- map.put(EdgeSocketCommonOptions.ENDPOINT.key(), "host:");
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
- Assertions.assertTrue(ex.getMessage().contains("expected format
host:port"));
- }
-
@Test
void blankEndpointTreatedAsNull() {
Map<String, Object> map = minimalConfig();
@@ -90,24 +54,6 @@ class EdgeSocketConfigTest {
Assertions.assertNull(config.getEndpoint());
}
- @Test
- void localQueueCapacityZeroThrows() {
- Map<String, Object> map = minimalConfig();
- map.put(EdgeSocketSourceOptions.LOCAL_QUEUE_CAPACITY.key(), 0);
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
-
Assertions.assertTrue(ex.getMessage().contains("local_queue_capacity"));
- }
-
- @Test
- void localQueueCapacityNegativeThrows() {
- Map<String, Object> map = minimalConfig();
- map.put(EdgeSocketSourceOptions.LOCAL_QUEUE_CAPACITY.key(), -5);
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
- Assertions.assertTrue(ex.getMessage().contains("greater than 0"));
- }
-
@Test
void validSecretKey32Bytes() {
byte[] key = "test-secret-key-32-bytes-aes256!".getBytes();
@@ -123,34 +69,6 @@ class EdgeSocketConfigTest {
Assertions.assertEquals(32, config.getSecretKeyBytes().length);
}
- @Test
- void secretKeyTooShortThrows() {
- byte[] shortKey = "only-16-bytes!!!".getBytes();
- Assertions.assertEquals(16, shortKey.length);
- String base64Key = Base64.getEncoder().encodeToString(shortKey);
-
- Map<String, Object> map = minimalConfig();
- map.put(EdgeSocketSourceOptions.PACKET_MODE.key(), "PACKET");
- map.put(EdgeSocketSourceOptions.SECRET_KEY.key(), base64Key);
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
- Assertions.assertTrue(ex.getMessage().contains("exactly 32 bytes"));
- }
-
- @Test
- void secretKeyTooLongThrows() {
- byte[] longKey =
"this-key-is-48-bytes-long-and-way-too-big-for-it".getBytes();
- Assertions.assertEquals(48, longKey.length);
- String base64Key = Base64.getEncoder().encodeToString(longKey);
-
- Map<String, Object> map = minimalConfig();
- map.put(EdgeSocketSourceOptions.PACKET_MODE.key(), "PACKET");
- map.put(EdgeSocketSourceOptions.SECRET_KEY.key(), base64Key);
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
- Assertions.assertTrue(ex.getMessage().contains("exactly 32 bytes"));
- }
-
@Test
void secretKeyIgnoredInRawMode() {
Map<String, Object> map = minimalConfig();
@@ -186,15 +104,6 @@ class EdgeSocketConfigTest {
Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
}
- @Test
- void missingTokenThrowsWhenAuthTypeIsToken() {
- Map<String, Object> map = minimalConfig();
- map.remove(EdgeSocketSourceOptions.TOKEN.key());
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
- Assertions.assertTrue(ex.getMessage().contains("token is required"));
- }
-
@Test
void defaultBackpressureOptions() {
EdgeSocketConfig config = createConfig(minimalConfig());
@@ -202,33 +111,6 @@ class EdgeSocketConfigTest {
Assertions.assertEquals(500, config.getQueueFullRetryAfterMs());
}
- @Test
- void invalidQueueBackpressureWatermarkRatioThrows() {
- Map<String, Object> map = minimalConfig();
-
map.put(EdgeSocketSourceOptions.QUEUE_BACKPRESSURE_WATERMARK_RATIO.key(), 1.5);
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
-
Assertions.assertTrue(ex.getMessage().contains("queue_backpressure_watermark_ratio"));
- }
-
- @Test
- void invalidQueueFullRetryAfterMsThrows() {
- Map<String, Object> map = minimalConfig();
- map.put(EdgeSocketSourceOptions.QUEUE_FULL_RETRY_AFTER_MS.key(), 0);
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
-
Assertions.assertTrue(ex.getMessage().contains("queue_full_retry_after_ms"));
- }
-
- @Test
- void blankTokenThrowsWhenAuthTypeIsToken() {
- Map<String, Object> map = minimalConfig();
- map.put(EdgeSocketSourceOptions.TOKEN.key(), " ");
- IllegalArgumentException ex =
- Assertions.assertThrows(IllegalArgumentException.class, () ->
createConfig(map));
- Assertions.assertTrue(ex.getMessage().contains("token is required"));
- }
-
private static Map<String, Object> minimalConfig() {
Map<String, Object> map = new HashMap<>();
map.put(EdgeSocketCommonOptions.PORT.key(), 9999);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/AbstractEdgeSocketIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/AbstractEdgeSocketIT.java
index ba43cdde96..60cbe7dec8 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/AbstractEdgeSocketIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/AbstractEdgeSocketIT.java
@@ -169,6 +169,56 @@ public abstract class AbstractEdgeSocketIT extends
TestSuiteBase implements Test
throw new IllegalStateException("Send records to edge socket timed
out");
}
+ protected List<Long>
sendRecordsThroughCollectorAndCollectWatermarks(List<String> messages)
+ throws Exception {
+ if (messages == null || messages.isEmpty()) {
+ throw new IllegalArgumentException("Messages should not be empty");
+ }
+ if (edgeSocketForwarderContainer == null) {
+ throw new IllegalStateException("Edge socket forwarder container
is not initialized");
+ }
+ String forwarderHost = edgeSocketForwarderContainer.getHost();
+ int forwarderPort =
edgeSocketForwarderContainer.getMappedPort(EDGE_FORWARDER_PORT);
+ long deadlineMillis = System.currentTimeMillis() +
TimeUnit.SECONDS.toMillis(60);
+ while (System.currentTimeMillis() < deadlineMillis) {
+ try (Socket socket = new Socket(forwarderHost, forwarderPort);
+ BufferedWriter writer =
+ new BufferedWriter(
+ new OutputStreamWriter(
+ socket.getOutputStream(),
StandardCharsets.UTF_8));
+ BufferedReader reader =
+ new BufferedReader(
+ new InputStreamReader(
+ socket.getInputStream(),
StandardCharsets.UTF_8))) {
+ socket.setSoTimeout(3000);
+ writeLine(writer, "__AUTH__:" + AUTH_TOKEN);
+ String authReply = readLine(reader);
+ Assertions.assertEquals("ACK", authReply, "Auth response
should be ACK");
+
+ List<Long> watermarks = new ArrayList<>();
+ long batchId = 1L;
+ for (String message : messages) {
+ sendMessageWithRetry(writer, reader, batchId, message);
+ watermarks.add(
+ awaitBatchCheckpointAckAndReturnWatermark(writer,
reader, batchId));
+ batchId++;
+ }
+ return watermarks;
+ } catch (SocketTimeoutException timeoutException) {
+ if (System.currentTimeMillis() >= deadlineMillis) {
+ throw timeoutException;
+ }
+ TimeUnit.MILLISECONDS.sleep(200);
+ } catch (IOException ioException) {
+ if (System.currentTimeMillis() >= deadlineMillis) {
+ throw ioException;
+ }
+ TimeUnit.MILLISECONDS.sleep(500);
+ }
+ }
+ throw new IllegalStateException("Send records to edge socket timed
out");
+ }
+
protected void awaitSinkContainsExpectedMessages(List<String>
expectedMessages) {
if (expectedMessages == null || expectedMessages.isEmpty()) {
throw new IllegalArgumentException("Expected messages should not
be empty");
@@ -379,6 +429,34 @@ public abstract class AbstractEdgeSocketIT extends
TestSuiteBase implements Test
"Timeout waiting checkpoint ACK for batch: " +
expectedBatchId);
}
+ private long awaitBatchCheckpointAckAndReturnWatermark(
+ BufferedWriter writer, BufferedReader reader, long
expectedBatchId) throws Exception {
+ long deadlineMillis = System.currentTimeMillis() +
TimeUnit.SECONDS.toMillis(30);
+ while (System.currentTimeMillis() < deadlineMillis) {
+ writeLine(writer, EDGE_COMMIT_PREFIX + expectedBatchId);
+ String reply = readLine(reader);
+ if ("PENDING".equals(reply)) {
+ TimeUnit.MILLISECONDS.sleep(200);
+ continue;
+ }
+ if (reply.startsWith("ACK:")) {
+ long ackedBatchId =
Long.parseLong(reply.substring("ACK:".length()));
+ if (ackedBatchId >= expectedBatchId) {
+ return ackedBatchId;
+ }
+ TimeUnit.MILLISECONDS.sleep(200);
+ continue;
+ }
+ if ("RETRY".equals(reply)) {
+ TimeUnit.MILLISECONDS.sleep(200);
+ continue;
+ }
+ throw new IllegalStateException("Unexpected commit response: " +
reply);
+ }
+ throw new IllegalStateException(
+ "Timeout waiting checkpoint ACK for batch: " +
expectedBatchId);
+ }
+
/**
* Send messages using PACKET mode framing (base64 + NONE compression, no
encryption). Each
* plain-text message is base64-encoded and wrapped in a JSON envelope
before sending.
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketMysql8_4IT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketMysql8_4IT.java
index 27d7b6d9d9..ba39d59c1a 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketMysql8_4IT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketMysql8_4IT.java
@@ -36,6 +36,7 @@ import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@@ -169,6 +170,42 @@ public class EdgeSocketMysql8_4IT extends
AbstractEdgeSocketIT {
}
}
+ @TestTemplate
+ public void testEdgeSocketCommitWatermarkAdvances(TestContainer container)
throws Exception {
+ String jobId = String.valueOf(JobIdGenerator.newJobId());
+ CompletableFuture<Container.ExecResult> jobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(
+ "/edge_socket_source_to_mysql84.conf",
jobId);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ try {
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
assertJobRunningOrSubmissionFailed(container, jobId, jobFuture));
+ ensureEdgeSocketForwarder();
+
+ List<String> sourceMessages = buildPlainTextMessages(3,
"watermark-value");
+ List<String> expectedMessages =
buildExpectedTransformedMessages(sourceMessages);
+ List<Long> watermarks =
sendRecordsThroughCollectorAndCollectWatermarks(sourceMessages);
+
+ Assertions.assertEquals(Arrays.asList(1L, 2L, 3L), watermarks);
+ awaitSinkContainsExpectedMessages(expectedMessages);
+ } finally {
+ Container.ExecResult cancelResult = container.cancelJob(jobId);
+ Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
+
+ Container.ExecResult jobResult = waitForJobResult(jobFuture);
+ Assertions.assertEquals(0, jobResult.getExitCode(),
jobResult.getStderr());
+ }
+ }
+
@TestTemplate
public void testEdgeSocketPacketModeToMysql84(TestContainer container)
throws Exception {
String jobId = String.valueOf(JobIdGenerator.newJobId());
