venkata91 commented on a change in pull request #30480:
URL: https://github.com/apache/spark/pull/30480#discussion_r602949966
##########
File path: core/src/main/scala/org/apache/spark/MapOutputTracker.scala
##########
@@ -449,21 +605,32 @@ private[spark] class MapOutputTrackerMaster(
try {
while (true) {
try {
- val data = mapOutputRequests.take()
- if (data == PoisonPill) {
+ val data = mapOutputTrackerMasterMessages.take()
+ if (data == PoisonPill) {
// Put PoisonPill back so that other MessageLoops can see it.
- mapOutputRequests.offer(PoisonPill)
+ mapOutputTrackerMasterMessages.offer(PoisonPill)
return
}
- val context = data.context
- val shuffleId = data.shuffleId
- val hostPort = context.senderAddress.hostPort
- logDebug("Handling request to send map output locations for
shuffle " + shuffleId +
- " to " + hostPort)
- val shuffleStatus = shuffleStatuses.get(shuffleId).head
- context.reply(
- shuffleStatus.serializedMapStatus(broadcastManager, isLocal,
minSizeForBroadcast,
- conf))
+
+ data match {
+ case GetMapStatusMessage(shuffleId, context) =>
+ val hostPort = context.senderAddress.hostPort
+ val shuffleStatus = shuffleStatuses.get(shuffleId).head
+ logDebug("Handling request to send map output locations for
shuffle " + shuffleId +
+ " to " + hostPort)
+ context.reply(
+ shuffleStatus.serializedOutputStatus(broadcastManager,
isLocal,
+ minSizeForBroadcast, conf, isMapOutput = true))
+
+ case GetMergeStatusMessage(shuffleId, context) =>
+ val hostPort = context.senderAddress.hostPort
+ val shuffleStatus = shuffleStatuses.get(shuffleId).head
+ logDebug("Handling request to send merge output locations for"
+
+ " shuffle " + shuffleId + " to " + hostPort)
+ context.reply(
+ shuffleStatus.serializedOutputStatus(broadcastManager,
isLocal,
+ minSizeForBroadcast, conf, isMapOutput = false))
+ }
Review comment:
As part of we would be adding one more message for getting shuffle
mergers https://issues.apache.org/jira/browse/SPARK-34826 which is why made
this change.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
