This is an automated email from the ASF dual-hosted git repository.
ethanfeng 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 d7798127 [CELEBORN-319] FlinkTransportClient should not reuse
connection. (#1252)
d7798127 is described below
commit d7798127c977f6f6eeb03fffc2cb72952ef0576a
Author: Ethan Feng <[email protected]>
AuthorDate: Tue Feb 21 11:16:30 2023 +0800
[CELEBORN-319] FlinkTransportClient should not reuse connection. (#1252)
---
.../plugin/flink/network/FlinkTransportClientFactory.java | 9 +++++----
.../apache/celeborn/plugin/flink/readclient/RssBufferStream.java | 7 ++++++-
.../apache/celeborn/plugin/flink/RemoteBufferStreamReader.java | 1 +
.../celeborn/common/network/client/TransportClientFactory.java | 2 +-
.../org/apache/celeborn/common/network/protocol/Message.java | 2 +-
5 files changed, 14 insertions(+), 7 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 61abcb6b..0472fb8e 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
@@ -18,6 +18,7 @@
package org.apache.celeborn.plugin.flink.network;
import java.io.IOException;
+import java.net.InetSocketAddress;
import java.util.function.Supplier;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
@@ -31,10 +32,10 @@ public class FlinkTransportClientFactory extends
TransportClientFactory {
super(context);
}
- public TransportClient createClient(
- String remoteHost, int remotePort, int partitionId, Supplier<ByteBuf>
supplier)
+ public TransportClient createClient(String remoteHost, int remotePort,
Supplier<ByteBuf> supplier)
throws IOException, InterruptedException {
- return createClient(
- remoteHost, remotePort, partitionId, new
TransportFrameDecoderWithBufferSupplier(supplier));
+ final InetSocketAddress resolvedAddress = new
InetSocketAddress(remoteHost, remotePort);
+ return internalCreateClient(
+ resolvedAddress, new
TransportFrameDecoderWithBufferSupplier(supplier));
}
}
diff --git
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
index 0be757c2..f5155aab 100644
---
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
+++
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
@@ -75,7 +75,6 @@ public class RssBufferStream {
clientFactory.createClient(
locations[currentLocationIndex].getHost(),
locations[currentLocationIndex].getFetchPort(),
- -1,
supplier);
OpenStreamWithCredit openBufferStream =
new OpenStreamWithCredit(
@@ -156,4 +155,10 @@ public class RssBufferStream {
public TransportClient getClient() {
return client;
}
+
+ public void close() {
+ if (client != null) {
+ client.close();
+ }
+ }
}
diff --git
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java
index 80f96942..d423f4d8 100644
---
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java
+++
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java
@@ -92,6 +92,7 @@ public class RemoteBufferStreamReader extends CreditListener {
public void close() {
isOpen = false;
client.getReadClientHandler().removeHandler(this.bufferStream.getStreamId());
+ bufferStream.close();
}
public boolean isOpened() {
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 de8e61de..4d5e7ffb 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
@@ -188,7 +188,7 @@ public class TransportClientFactory implements Closeable {
*
* <p>As with {@link #createClient(String, int)}, this method is blocking.
*/
- private TransportClient internalCreateClient(
+ protected TransportClient internalCreateClient(
InetSocketAddress address, ChannelInboundHandlerAdapter decoder)
throws IOException, InterruptedException {
Bootstrap bootstrap = new Bootstrap();
diff --git
a/common/src/main/java/org/apache/celeborn/common/network/protocol/Message.java
b/common/src/main/java/org/apache/celeborn/common/network/protocol/Message.java
index d2bbd55a..4129ccb8 100644
---
a/common/src/main/java/org/apache/celeborn/common/network/protocol/Message.java
+++
b/common/src/main/java/org/apache/celeborn/common/network/protocol/Message.java
@@ -56,7 +56,7 @@ public abstract class Message implements Encodable {
/** Whether the body should be copied out in frame decoder. */
public boolean needCopyOut() {
- return false;
+ return true;
}
protected boolean equals(Message other) {
