mmodzelewski commented on code in PR #3657:
URL: https://github.com/apache/iggy/pull/3657#discussion_r3637361681
##########
foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java:
##########
@@ -180,9 +179,25 @@ public void flush(boolean endOfInput) throws IOException {
@Override
public void close() throws Exception {
- // Flush any remaining buffered records
- flush(true);
- // Note: HTTP client doesn't have close() method - connections managed
by Java HttpClient pool
+ IOException flushException = null;
+ try {
+ flush(true);
+ } catch (IOException e) {
+ flushException = e;
+ }
Review Comment:
`flush()` wraps send failures in `ConnectorException`, which extends
`RuntimeException`, so this `catch (IOException e)` never fires for the most
common failure mode. The exception then propagates before `tcpClient.close()`
runs, leaking the Netty channels and event-loop threads in the TaskManager JVM.
This defeats the scenario this block was written for. Widening the catch keeps
the intended semantics:
```suggestion
Exception flushException = null;
try {
flush(true);
} catch (Exception e) {
flushException = e;
}
```
##########
foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySource.java:
##########
@@ -154,23 +155,12 @@ public
SimpleVersionedSerializer<IggySourceEnumeratorState> getEnumeratorCheckpo
*/
private AsyncIggyTcpClient createAsyncIggyClient() {
try {
- // Parse host and port from server address
- String serverAddress = connectionConfig.getServerAddress();
- String host;
- int port = 8090; // Default TCP port
-
- if (serverAddress.contains(":")) {
- String[] parts = serverAddress.split(":");
- host = parts[0];
- port = Integer.parseInt(parts[1]);
- } else {
- host = serverAddress;
- }
+ TcpEndpoint endpoint =
TcpEndpoint.parse(connectionConfig.getServerAddress());
// Create async TCP client using builder pattern with auto connect
and login
return AsyncIggyTcpClient.builder()
- .host(host)
- .port(port)
+ .host(endpoint.host())
+ .port(endpoint.port())
.credentials(connectionConfig.getUsername(),
connectionConfig.getPassword())
.connectionPoolSize(4)
.buildAndLogin()
Review Comment:
The sink now honors connection/request timeouts, retry policy, and TLS from
`IggyConnectionConfig`, but the source still ignores them. With
`enableTls(true)` the sink connects over TLS while the source silently uses
plaintext, so the same config still behaves differently between source and
sink. `AsyncIggyTcpClientBuilder` already exposes the same options:
```suggestion
return AsyncIggyTcpClient.builder()
.host(endpoint.host())
.port(endpoint.port())
.credentials(connectionConfig.getUsername(),
connectionConfig.getPassword())
.connectionTimeout(connectionConfig.getConnectionTimeout())
.requestTimeout(connectionConfig.getRequestTimeout())
.retryPolicy(RetryPolicy.fixedDelay(
connectionConfig.getMaxRetries(),
connectionConfig.getRetryBackoff()))
.tls(connectionConfig.isEnableTls())
.connectionPoolSize(4)
.buildAndLogin()
```
(requires an `org.apache.iggy.config.RetryPolicy` import)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
