Skip to content

Commit 6e8359a

Browse files
Standing-ManStanding-Man
committed
refactor(connector): unify TCP address parsing
Signed-off-by: StandingMan <jmtangcs@gmail.com>
1 parent de5432c commit 6e8359a

6 files changed

Lines changed: 178 additions & 48 deletions

File tree

‎foreign/java/.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
.project
33
.settings/
44
.gradle/
5+
.kotlin/
56
build/
67
out/
78

‎foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/config/TcpEndpoint.java‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iggy.connector.config;
21+
22+
import java.net.URI;
23+
import java.net.URISyntaxException;
24+
25+
/**
26+
* Host and port of an Iggy TCP endpoint.
27+
*
28+
* @param host server host
29+
* @param port server TCP port
30+
*/
31+
public record TcpEndpoint(String host, int port) {
32+
33+
public static final int DEFAULT_PORT = 8090;
34+
35+
/**
36+
* Parses an Iggy TCP server address.
37+
*
38+
* @param serverAddress address in host, host:port, or tcp://host:port form
39+
* @return parsed TCP endpoint
40+
*/
41+
public static TcpEndpoint parse(String serverAddress) {
42+
if (serverAddress == null || serverAddress.isBlank()) {
43+
throw new IllegalArgumentException("TCP server address cannot be null or blank");
44+
}
45+
46+
URI uri = parseUri(serverAddress);
47+
validateScheme(uri);
48+
String host = extractHost(uri, serverAddress);
49+
validateComponents(uri, serverAddress);
50+
51+
int port = uri.getPort() >= 0 ? uri.getPort() : DEFAULT_PORT;
52+
return new TcpEndpoint(host, port);
53+
}
54+
55+
private static URI parseUri(String serverAddress) {
56+
try {
57+
return serverAddress.contains("://") ? new URI(serverAddress) : new URI("tcp://" + serverAddress);
58+
} catch (URISyntaxException e) {
59+
throw new IllegalArgumentException("Invalid TCP server address: " + serverAddress, e);
60+
}
61+
}
62+
63+
private static void validateScheme(URI uri) {
64+
if (!"tcp".equalsIgnoreCase(uri.getScheme())) {
65+
throw new IllegalArgumentException("Unsupported TCP server address scheme: " + uri.getScheme());
66+
}
67+
}
68+
69+
private static String extractHost(URI uri, String serverAddress) {
70+
String host = uri.getHost();
71+
if (host == null || host.isBlank()) {
72+
throw new IllegalArgumentException("Cannot extract host from TCP server address: " + serverAddress);
73+
}
74+
return host;
75+
}
76+
77+
private static void validateComponents(URI uri, String serverAddress) {
78+
if (uri.getUserInfo() != null
79+
|| !uri.getPath().isEmpty()
80+
|| uri.getQuery() != null
81+
|| uri.getFragment() != null) {
82+
throw new IllegalArgumentException("Invalid TCP server address: " + serverAddress);
83+
}
84+
}
85+
}

‎foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySink.java‎

Lines changed: 23 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -25,31 +25,32 @@
2525
import org.apache.iggy.client.blocking.tcp.IggyTcpClient;
2626
import org.apache.iggy.config.RetryPolicy;
2727
import org.apache.iggy.connector.config.IggyConnectionConfig;
28+
import org.apache.iggy.connector.config.TcpEndpoint;
2829
import org.apache.iggy.connector.serialization.SerializationSchema;
2930

3031
import java.io.IOException;
3132
import java.io.Serializable;
32-
import java.net.URI;
33-
import java.net.URISyntaxException;
3433
import java.time.Duration;
3534

3635
/**
3736
* Flink Sink implementation for writing to Iggy streams.
3837
* Implements the Flink Sink V2 API for integration with DataStream API.
3938
*
40-
* <p>Example usage:
39+
* <p>
40+
* Example usage:
41+
*
4142
* <pre>{@code
4243
* events.sinkTo(
43-
* IggySink.<Event>builder()
44-
* .setConnectionConfig(connectionConfig)
45-
* .setStreamId("my-stream")
46-
* .setTopicId("my-topic")
47-
* .setSerializer(new JsonSerializationSchema<>())
48-
* .setBatchSize(100)
49-
* .setFlushInterval(Duration.ofSeconds(5))
50-
* .withBalancedPartitioning()
51-
* .build()
52-
* ).name("Iggy Sink");
44+
* IggySink.<Event>builder()
45+
* .setConnectionConfig(connectionConfig)
46+
* .setStreamId("my-stream")
47+
* .setTopicId("my-topic")
48+
* .setSerializer(new JsonSerializationSchema<>())
49+
* .setBatchSize(100)
50+
* .setFlushInterval(Duration.ofSeconds(5))
51+
* .withBalancedPartitioning()
52+
* .build())
53+
* .name("Iggy Sink");
5354
* }</pre>
5455
*
5556
* @param <T> the type of records to write
@@ -70,12 +71,12 @@ public class IggySink<T> implements Sink<T>, Serializable {
7071
* Creates a new Iggy sink.
7172
* Use {@link #builder()} to construct instances.
7273
*
73-
* @param connectionConfig the connection configuration
74-
* @param streamId the stream identifier
75-
* @param topicId the topic identifier
76-
* @param serializer the serialization schema
77-
* @param batchSize the batch size for buffering
78-
* @param flushInterval the maximum flush interval
74+
* @param connectionConfig the connection configuration
75+
* @param streamId the stream identifier
76+
* @param topicId the topic identifier
77+
* @param serializer the serialization schema
78+
* @param batchSize the batch size for buffering
79+
* @param flushInterval the maximum flush interval
7980
* @param partitioningStrategy the partitioning strategy
8081
*/
8182
public IggySink(
@@ -120,28 +121,18 @@ public SinkWriter<T> createWriter(WriterInitContext context) throws IOException
120121
*/
121122
private IggyTcpClient createTcpClient() {
122123
try {
123-
String serverAddress = connectionConfig.getServerAddress();
124-
URI uri = serverAddress.contains("://") ? new URI(serverAddress) : new URI("tcp://" + serverAddress);
125-
126-
String host = uri.getHost();
127-
if (host == null) {
128-
throw new IllegalArgumentException("Cannot extract host from: " + serverAddress);
129-
}
130-
int port = uri.getPort() >= 0 ? uri.getPort() : 8090;
124+
TcpEndpoint endpoint = TcpEndpoint.parse(connectionConfig.getServerAddress());
131125

132126
return IggyTcpClient.builder()
133-
.host(host)
134-
.port(port)
127+
.host(endpoint.host())
128+
.port(endpoint.port())
135129
.credentials(connectionConfig.getUsername(), connectionConfig.getPassword())
136130
.connectionTimeout(connectionConfig.getConnectionTimeout())
137131
.requestTimeout(connectionConfig.getRequestTimeout())
138132
.retryPolicy(RetryPolicy.fixedDelay(
139133
connectionConfig.getMaxRetries(), connectionConfig.getRetryBackoff()))
140134
.tls(connectionConfig.isEnableTls())
141135
.buildAndLogin();
142-
143-
} catch (URISyntaxException e) {
144-
throw new RuntimeException("Invalid server address format: " + connectionConfig.getServerAddress(), e);
145136
} catch (RuntimeException e) {
146137
throw new RuntimeException("Failed to create TCP Iggy client", e);
147138
}

‎foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -179,10 +179,10 @@ public void flush(boolean endOfInput) throws IOException {
179179

180180
@Override
181181
public void close() throws Exception {
182-
Exception flushException = null;
182+
IOException flushException = null;
183183
try {
184184
flush(true);
185-
} catch (Exception e) {
185+
} catch (IOException e) {
186186
flushException = e;
187187
}
188188

‎foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySource.java‎

Lines changed: 4 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
3030
import org.apache.iggy.connector.config.IggyConnectionConfig;
3131
import org.apache.iggy.connector.config.OffsetConfig;
32+
import org.apache.iggy.connector.config.TcpEndpoint;
3233
import org.apache.iggy.consumergroup.Consumer;
3334

3435
import java.io.Serializable;
@@ -154,23 +155,12 @@ public SimpleVersionedSerializer<IggySourceEnumeratorState> getEnumeratorCheckpo
154155
*/
155156
private AsyncIggyTcpClient createAsyncIggyClient() {
156157
try {
157-
// Parse host and port from server address
158-
String serverAddress = connectionConfig.getServerAddress();
159-
String host;
160-
int port = 8090; // Default TCP port
161-
162-
if (serverAddress.contains(":")) {
163-
String[] parts = serverAddress.split(":");
164-
host = parts[0];
165-
port = Integer.parseInt(parts[1]);
166-
} else {
167-
host = serverAddress;
168-
}
158+
TcpEndpoint endpoint = TcpEndpoint.parse(connectionConfig.getServerAddress());
169159

170160
// Create async TCP client using builder pattern with auto connect and login
171161
return AsyncIggyTcpClient.builder()
172-
.host(host)
173-
.port(port)
162+
.host(endpoint.host())
163+
.port(endpoint.port())
174164
.credentials(connectionConfig.getUsername(), connectionConfig.getPassword())
175165
.connectionPoolSize(4)
176166
.buildAndLogin()

‎foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/config/TcpEndpointTest.java‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iggy.connector.config;
21+
22+
import org.junit.jupiter.api.Test;
23+
24+
import static org.assertj.core.api.Assertions.assertThat;
25+
import static org.assertj.core.api.Assertions.assertThatThrownBy;
26+
27+
class TcpEndpointTest {
28+
29+
@Test
30+
void shouldUseDefaultPortForHost() {
31+
assertThat(TcpEndpoint.parse("localhost")).isEqualTo(new TcpEndpoint("localhost", 8090));
32+
}
33+
34+
@Test
35+
void shouldParseHostAndPort() {
36+
assertThat(TcpEndpoint.parse("iggy:8091")).isEqualTo(new TcpEndpoint("iggy", 8091));
37+
}
38+
39+
@Test
40+
void shouldParseTcpUri() {
41+
assertThat(TcpEndpoint.parse("tcp://iggy.example.com:8092"))
42+
.isEqualTo(new TcpEndpoint("iggy.example.com", 8092));
43+
}
44+
45+
@Test
46+
void shouldParseIpv6Address() {
47+
assertThat(TcpEndpoint.parse("tcp://[::1]:8093")).isEqualTo(new TcpEndpoint("[::1]", 8093));
48+
}
49+
50+
@Test
51+
void shouldRejectUnsupportedScheme() {
52+
assertThatThrownBy(() -> TcpEndpoint.parse("http://localhost:3000"))
53+
.isInstanceOf(IllegalArgumentException.class)
54+
.hasMessageContaining("Unsupported TCP server address scheme");
55+
}
56+
57+
@Test
58+
void shouldRejectBlankAddress() {
59+
assertThatThrownBy(() -> TcpEndpoint.parse(" "))
60+
.isInstanceOf(IllegalArgumentException.class)
61+
.hasMessageContaining("cannot be null or blank");
62+
}
63+
}

0 commit comments

Comments
 (0)