venkata91 commented on a change in pull request #30480:
URL: https://github.com/apache/spark/pull/30480#discussion_r612785890
##########
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:
Just realized I totally misunderstood the first comment above @mridulm
--
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:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
