RexXiong commented on code in PR #973:
URL:
https://github.com/apache/incubator-celeborn/pull/973#discussion_r1023481662
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java:
##########
@@ -257,13 +259,58 @@ private String genAddressPair(PartitionLocation loc) {
private ConcurrentHashMap<Integer, PartitionLocation> registerShuffle(
String appId, int shuffleId, int numMappers, int numPartitions) {
+ return registerShuffle(
+ shuffleId,
+ numMappers,
+ numMappers,
+ () ->
+ driverRssMetaService.askSync(
+ RegisterShuffle$.MODULE$.apply(appId, shuffleId, numMappers,
numPartitions),
+ ClassTag$.MODULE$.apply(PbRegisterShuffleResponse.class)));
+ }
+
+ @Override
+ public PartitionLocation registerMapPartitionTask(
+ String appId, int shuffleId, int numMappers, int mapId, int attemptId) {
+ int partitionId = PackedPartitionId.packedPartitionId(mapId, attemptId);
+ logger.info(
+ "register mapPartitionTask, mapId: {}, attemptId: {}, partitionId: {}",
+ mapId,
+ attemptId,
+ partitionId);
+ if (attemptId == 0) {
+ return registerMapPartitionTaskWithFirstAttempt(
+ appId, shuffleId, numMappers, mapId, attemptId, partitionId);
+ }
+
+ // TODO
+ throw new UnsupportedOperationException("can not register shuffle task
with attempt beyond 0");
+ }
+
+ private PartitionLocation registerMapPartitionTaskWithFirstAttempt(
+ String appId, int shuffleId, int numMappers, int mapId, int attemptId,
int partitionId) {
+ ConcurrentHashMap<Integer, PartitionLocation> partitionLocationMap =
+ registerShuffle(
+ shuffleId,
+ numMappers,
+ numMappers,
+ () ->
+ driverRssMetaService.askSync(
+ RegisterMapPartitionTask$.MODULE$.apply(
+ appId, shuffleId, numMappers, mapId, attemptId,
partitionId),
+ ClassTag$.MODULE$.apply(PbRegisterShuffleResponse.class)));
+ return partitionLocationMap.get(partitionId);
+ }
+
+ public ConcurrentHashMap<Integer, PartitionLocation> registerShuffle(
+ int shuffleId,
+ int numMappers,
+ int numPartitions,
+ Callable<PbRegisterShuffleResponse> callable) {
Review Comment:
this registerShuffleInternal function will response for both registerShuffle
and registerMapPartitionTask shuffle, So Callable is needed
--
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]
