You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2023/03/30 03:44:25 UTC
[incubator-celeborn] branch main updated: [CELEBORN-489][FLINK]fix retry client for open stream (#1397)
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 45013b8ba [CELEBORN-489][FLINK]fix retry client for open stream (#1397)
45013b8ba is described below
commit 45013b8baea5616beeba814057dbfd09a53f04fe
Author: Shuang <lv...@gmail.com>
AuthorDate: Thu Mar 30 11:44:19 2023 +0800
[CELEBORN-489][FLINK]fix retry client for open stream (#1397)
---
.../flink/network/FlinkTransportClientFactory.java | 27 +++++++++++++++++++---
.../flink/readclient/FlinkShuffleClientImpl.java | 3 ++-
.../network/client/TransportClientFactory.java | 11 ---------
3 files changed, 26 insertions(+), 15 deletions(-)
diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/FlinkTransportClientFactory.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/FlinkTransportClientFactory.java
index 812b4c12a..5492c9505 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/FlinkTransportClientFactory.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/FlinkTransportClientFactory.java
@@ -36,16 +36,37 @@ public class FlinkTransportClientFactory extends TransportClientFactory {
public static final Logger logger = LoggerFactory.getLogger(FlinkTransportClientFactory.class);
private ConcurrentHashMap<Long, Supplier<ByteBuf>> bufferSuppliers;
+ private final int fetchMaxRetries;
- public FlinkTransportClientFactory(TransportContext context) {
+ public FlinkTransportClientFactory(TransportContext context, int fetchMaxRetries) {
super(context);
bufferSuppliers = JavaUtils.newConcurrentHashMap();
+ this.fetchMaxRetries = fetchMaxRetries;
}
public TransportClient createClientWithRetry(String remoteHost, int remotePort)
throws IOException, InterruptedException {
- return createClientWithRetry(
- remoteHost, remotePort, -1, new TransportFrameDecoderWithBufferSupplier(bufferSuppliers));
+ int retryCount = fetchMaxRetries;
+
+ while (retryCount > 0) {
+ try {
+ return createClient(remoteHost, remotePort);
+ } catch (IOException e) {
+ retryCount--;
+ logger.warn(
+ "Retrying ({}/{}) times create client to {}:{}",
+ retryCount,
+ fetchMaxRetries,
+ remoteHost,
+ remotePort,
+ e);
+ if (retryCount == 0) {
+ throw e;
+ }
+ }
+ }
+
+ return null;
}
public TransportClient createClient(String remoteHost, int remotePort)
diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java
index 321dcec15..d71d440a8 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java
@@ -105,7 +105,8 @@ public class FlinkShuffleClientImpl extends ShuffleClientImpl {
TransportContext context =
new TransportContext(
dataTransportConf, readClientHandler, conf.clientCloseIdleConnections());
- this.flinkTransportClientFactory = new FlinkTransportClientFactory(context);
+ this.flinkTransportClientFactory =
+ new FlinkTransportClientFactory(context, conf.fetchMaxRetries());
}
public RssBufferStream readBufferedPartition(
diff --git a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java
index f8de07d67..fece7cb67 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java
@@ -116,17 +116,6 @@ public class TransportClientFactory implements Closeable {
return createClient(remoteHost, remotePort, partitionId, new TransportFrameDecoder());
}
- public TransportClient createClientWithRetry(
- String remoteHost, int remotePort, int partitionId, ChannelInboundHandlerAdapter decoder)
- throws IOException, InterruptedException {
- try {
- return createClient(remoteHost, remotePort, partitionId, decoder);
- } catch (IOException | InterruptedException e) {
- logger.warn("Retrying create client to {}:{}", remoteHost, remotePort, e);
- return createClient(remoteHost, remotePort, partitionId, decoder);
- }
- }
-
public TransportClient createClient(
String remoteHost, int remotePort, int partitionId, ChannelInboundHandlerAdapter decoder)
throws IOException, InterruptedException {