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/06/20 12:17:22 UTC
[incubator-celeborn] branch branch-0.3 updated: [CELEBORN-700][COMPATIBILITY] Fix compatibility issue caused by WorkerInfo
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/branch-0.3 by this push:
new 47954a4ad [CELEBORN-700][COMPATIBILITY] Fix compatibility issue caused by WorkerInfo
47954a4ad is described below
commit 47954a4adda603cf78bbe50cbb59007e086da222
Author: zky.zhoukeyong <zk...@alibaba-inc.com>
AuthorDate: Tue Jun 20 20:17:00 2023 +0800
[CELEBORN-700][COMPATIBILITY] Fix compatibility issue caused by WorkerInfo
### What changes were proposed in this pull request?
Fixes compatibility issue introduced by change of WorkerInfo.
### Why are the changes needed?
When testing with branch-0.2 client and main server, I got the following error:
```
Caused by: scala.MatchError: [Ljava.lang.String;414ca35a (of class [Ljava.lang.String;)
at org.apache.celeborn.common.util.PbSerDeUtils$.$anonfun$fromPbWorkerResource$1(PbSerDeUtils.scala:298)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at org.apache.celeborn.common.util.PbSerDeUtils$.fromPbWorkerResource(PbSerDeUtils.scala:297)
at org.apache.celeborn.common.protocol.message.ControlMessages$.fromTransportMessage(ControlMessages.scala:863)
at org.apache.celeborn.common.util.Utils$.fromTransportMessage(Utils.scala:828)
at org.apache.celeborn.common.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:110)
at org.apache.celeborn.common.rpc.netty.NettyRpcEnv.$anonfun$deserialize$2(NettyRpcEnv.scala:276)
at scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)
at org.apache.celeborn.common.rpc.netty.NettyRpcEnv.deserialize(NettyRpcEnv.scala:323)
at org.apache.celeborn.common.rpc.netty.NettyRpcEnv.$anonfun$deserialize$1(NettyRpcEnv.scala:275)
at scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)
at org.apache.celeborn.common.rpc.netty.NettyRpcEnv.deserialize(NettyRpcEnv.scala:275)
at org.apache.celeborn.common.rpc.netty.NettyRpcEnv.$anonfun$ask$6(NettyRpcEnv.scala:235)
at org.apache.celeborn.common.rpc.netty.NettyRpcEnv.$anonfun$ask$6$adapted(NettyRpcEnv.scala:235)
at org.apache.celeborn.common.rpc.netty.RpcOutboxMessage.onSuccess(Outbox.scala:82)
at org.apache.celeborn.common.network.client.TransportResponseHandler.handle(TransportResponseHandler.java:180)
at org.apache.celeborn.common.network.server.TransportChannelHandler.channelRead(TransportChannelHandler.java:119)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.celeborn.shaded.io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.apache.celeborn.common.network.util.TransportFrameDecoder.channelRead(TransportFrameDecoder.java:74)
```
And this is introduced by https://github.com/apache/incubator-celeborn/commit/811e192bbd2da279d145bec62f51b5d87e7f68b5#diff-b61712c3683306f65cd2ca051b54075952897a899951f3e37ec3968e7ba75710
### Does this PR introduce _any_ user-facing change?
Yes, it fixes compatibility error when using branch-0.2 client and main server.
### How was this patch tested?
Manual test.
Closes #1610 from waitinfuture/700.
Authored-by: zky.zhoukeyong <zk...@alibaba-inc.com>
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
(cherry picked from commit fdb126112ea77e3c06bef54762cf814f0ba4ed26)
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
common/src/main/proto/TransportMessages.proto | 1 +
.../org/apache/celeborn/common/meta/WorkerInfo.scala | 18 ------------------
.../org/apache/celeborn/common/util/PbSerDeUtils.scala | 15 +++++++++++----
3 files changed, 12 insertions(+), 22 deletions(-)
diff --git a/common/src/main/proto/TransportMessages.proto b/common/src/main/proto/TransportMessages.proto
index eaf1cb705..31c201e65 100644
--- a/common/src/main/proto/TransportMessages.proto
+++ b/common/src/main/proto/TransportMessages.proto
@@ -98,6 +98,7 @@ message PbPartitionLocation {
message PbWorkerResource {
repeated PbPartitionLocation masterPartitions = 1;
repeated PbPartitionLocation slavePartitions = 2;
+ string networkLocation = 3;
}
message PbDiskInfo {
diff --git a/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala b/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala
index b1bf51cd8..78b729eb8 100644
--- a/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala
@@ -159,10 +159,6 @@ class WorkerInfo(
s"$host:$rpcPort:$pushPort:$fetchPort:$replicatePort"
}
- def toInfoId(): String = {
- s"$host:$rpcPort:$pushPort:$fetchPort:$replicatePort:$networkLocation"
- }
-
def slotAvailable(): Boolean = this.synchronized {
diskInfos.asScala.exists { case (_, disk) => (disk.maxSlots - disk.activeSlots) > 0 }
}
@@ -279,18 +275,4 @@ object WorkerInfo {
val Array(host, rpcPort, pushPort, fetchPort, replicatePort) = id.split(":")
new WorkerInfo(host, rpcPort.toInt, pushPort.toInt, fetchPort.toInt, replicatePort.toInt)
}
-
- def fromInfoId(id: String): WorkerInfo = {
- val infoArr = id.split(":")
- if (infoArr.length == 6) {
- val Array(host, rpcPort, pushPort, fetchPort, replicatePort, networkLocation) = id.split(":")
- val workerInfo =
- new WorkerInfo(host, rpcPort.toInt, pushPort.toInt, fetchPort.toInt, replicatePort.toInt)
- workerInfo.networkLocation = networkLocation
- workerInfo
- } else {
- val Array(host, rpcPort, pushPort, fetchPort, replicatePort) = id.split(":")
- new WorkerInfo(host, rpcPort.toInt, pushPort.toInt, fetchPort.toInt, replicatePort.toInt)
- }
- }
}
diff --git a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
index be4aab245..9841a56ec 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
@@ -293,12 +293,17 @@ object PbSerDeUtils {
def fromPbWorkerResource(pbWorkerResource: util.Map[String, PbWorkerResource]): WorkerResource = {
val slots = new WorkerResource()
- pbWorkerResource.asScala.foreach { case (infoId, pbWorkerResource) =>
+ pbWorkerResource.asScala.foreach { case (uniqueId, pbWorkerResource) =>
+ val Array(host, rpcPort, pushPort, fetchPort, replicatePort) = uniqueId.split(":")
+ val networkLocation = pbWorkerResource.getNetworkLocation
+ val workerInfo =
+ new WorkerInfo(host, rpcPort.toInt, pushPort.toInt, fetchPort.toInt, replicatePort.toInt)
+ workerInfo.networkLocation = networkLocation
val masterPartitionLocation = new util.ArrayList[PartitionLocation](pbWorkerResource
.getMasterPartitionsList.asScala.map(PbSerDeUtils.fromPbPartitionLocation).asJava)
val slavePartitionLocation = new util.ArrayList[PartitionLocation](pbWorkerResource
.getSlavePartitionsList.asScala.map(PbSerDeUtils.fromPbPartitionLocation).asJava)
- slots.put(WorkerInfo.fromInfoId(infoId), (masterPartitionLocation, slavePartitionLocation))
+ slots.put(workerInfo, (masterPartitionLocation, slavePartitionLocation))
}
slots
}
@@ -309,8 +314,10 @@ object PbSerDeUtils {
val slavePartitions = slaveLocations.asScala.map(PbSerDeUtils.toPbPartitionLocation).asJava
val pbWorkerResource = PbWorkerResource.newBuilder()
.addAllMasterPartitions(masterPartitions)
- .addAllSlavePartitions(slavePartitions).build()
- workerInfo.toInfoId() -> pbWorkerResource
+ .addAllSlavePartitions(slavePartitions)
+ .setNetworkLocation(workerInfo.networkLocation)
+ .build()
+ workerInfo.toUniqueId() -> pbWorkerResource
}.asJava
}