This is an automated email from the ASF dual-hosted git repository.
steveloughran pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new bc15cd3e0cf MAPREDUCE-6706. TaskUmbilicalProtocol to use
ProtobufRPCEngine2 (#8422) (#8574)
bc15cd3e0cf is described below
commit bc15cd3e0cfd64532e0ad2edc3b7d285e3e6fdd2
Author: Steve Loughran <[email protected]>
AuthorDate: Tue Jun 30 09:47:23 2026 +0100
MAPREDUCE-6706. TaskUmbilicalProtocol to use ProtobufRPCEngine2 (#8422)
(#8574)
This moves the TaskUmbilicalProtocol between Mappers, Reducers and the
MRAppMaster to using ProtobufRPCEngine2
Contains code written by Github Copilot
Contributed by Steve Loughran
---
.../hadoop-mapreduce-client-app/pom.xml | 31 ++
.../hadoop/mapred/LocalContainerLauncher.java | 72 +++--
.../hadoop/mapred/TaskAttemptListenerImpl.java | 107 ++++---
.../java/org/apache/hadoop/mapred/YarnChild.java | 43 +--
.../mapred/protocolPB/TaskUmbilicalProtocolPB.java | 39 +++
.../TaskUmbilicalProtocolPBClientImpl.java | 337 +++++++++++++++++++++
...askUmbilicalProtocolServerSideTranslatorPB.java | 322 ++++++++++++++++++++
.../protocolPB/TaskUmbilicalProtocolUtils.java | 190 ++++++++++++
.../hadoop/mapred/protocolPB/package-info.java} | 51 +---
.../app/security/authorize/MRAMPolicyProvider.java | 4 +-
.../src/main/proto/TaskUmbilicalProtocol.proto | 187 ++++++++++++
.../java/org/apache/hadoop/mapred/JvmContext.java | 6 +-
.../org/apache/hadoop/mapred/SortedRanges.java | 6 +-
.../java/org/apache/hadoop/mapred/TaskStatus.java | 46 ++-
.../TestUmbilicalProtocolWithJobToken.java | 48 +--
15 files changed, 1321 insertions(+), 168 deletions(-)
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/pom.xml
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/pom.xml
index ed611676968..f4c0d70a01d 100644
---
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/pom.xml
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/pom.xml
@@ -192,6 +192,37 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.xolstice.maven.plugins</groupId>
+ <artifactId>protobuf-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>src-compile-protoc</id>
+ <configuration>
+ <skip>false</skip>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>com.google.code.maven-replacer-plugin</groupId>
+ <artifactId>replacer</artifactId>
+ <executions>
+ <execution>
+ <id>replace-generated-sources</id>
+ <configuration>
+ <skip>false</skip>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <configuration>
+
<excludePackageNames>org.apache.hadoop.mapred.proto</excludePackageNames>
+ </configuration>
+ </plugin>
</plugins>
</build>
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/LocalContainerLauncher.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/LocalContainerLauncher.java
index 1548bcc3c6b..11dc9461fff 100644
---
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/LocalContainerLauncher.java
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/LocalContainerLauncher.java
@@ -93,7 +93,7 @@ public class LocalContainerLauncher extends AbstractService
implements
private Thread eventHandler;
private byte[] encryptedSpillKey = new byte[] {0};
private BlockingQueue<ContainerLauncherEvent> eventQueue =
- new LinkedBlockingQueue<ContainerLauncherEvent>();
+ new LinkedBlockingQueue<>();
public LocalContainerLauncher(AppContext context,
TaskUmbilicalProtocol umbilical) {
@@ -115,8 +115,8 @@ public LocalContainerLauncher(AppContext context,
try {
curFC = FileContext.getFileContext(curDir.toURI());
} catch (UnsupportedFileSystemException ufse) {
- LOG.error("Local filesystem " + curDir.toURI().toString()
- + " is unsupported?? (should never happen)");
+ LOG.error("Local filesystem {} is unsupported?? (should never happen)",
+ curDir.toURI());
}
// Save list of files/dirs that are supposed to be present so can delete
@@ -126,10 +126,8 @@ public LocalContainerLauncher(AppContext context,
// uberization in order to run correctly).
File[] curLocalFiles = curDir.listFiles();
if (curLocalFiles != null) {
- HashSet<File> lf = new HashSet<File>(curLocalFiles.length);
- for (int j = 0; j < curLocalFiles.length; ++j) {
- lf.add(curLocalFiles[j]);
- }
+ HashSet<File> lf = new HashSet<>(curLocalFiles.length);
+ Collections.addAll(lf, curLocalFiles);
localizedFiles = Collections.unmodifiableSet(lf);
}
@@ -157,26 +155,29 @@ public void serviceStart() throws Exception {
// thread context classloader so that it can be used by the event handler
// as well as the subtask runner threads
if (jobClassLoader != null) {
- LOG.info("Setting " + jobClassLoader +
- " as the context classloader of thread " + eventHandler.getName());
+ LOG.info("Setting {} as the context classloader of thread {}",
jobClassLoader,
+ eventHandler.getName());
eventHandler.setContextClassLoader(jobClassLoader);
} else {
// note the current TCCL
- LOG.info("Context classloader of thread " + eventHandler.getName() +
- ": " + eventHandler.getContextClassLoader());
+ LOG.info("Context classloader of thread {}: {}", eventHandler.getName(),
+ eventHandler.getContextClassLoader());
}
eventHandler.start();
super.serviceStart();
}
public void serviceStop() throws Exception {
- if (eventHandler != null) {
- eventHandler.interrupt();
- }
- if (taskRunner != null) {
- taskRunner.shutdownNow();
+ try {
+ if (eventHandler != null) {
+ eventHandler.interrupt();
+ }
+ if (taskRunner != null) {
+ taskRunner.shutdownNow();
+ }
+ } finally {
+ super.serviceStop();
}
- super.serviceStop();
}
@Override
@@ -223,7 +224,7 @@ private class EventHandler implements Runnable {
private int finishedSubMaps = 0;
private final Map<TaskAttemptId,Future<?>> futures =
- new ConcurrentHashMap<TaskAttemptId,Future<?>>();
+ new ConcurrentHashMap<>();
EventHandler() {
}
@@ -235,7 +236,7 @@ public void run() {
// Collect locations of map outputs to give to reduces
final Map<TaskAttemptID, MapOutputFile> localMapFiles =
- new HashMap<TaskAttemptID, MapOutputFile>();
+ new HashMap<>();
// _must_ either run subtasks sequentially or accept expense of new JVMs
// (i.e., fork()), else will get weird failures when maps try to create/
@@ -244,11 +245,11 @@ public void run() {
try {
event = eventQueue.take();
} catch (InterruptedException e) { // mostly via T_KILL? JOB_KILL?
- LOG.warn("Returning, interrupted : " + e);
+ LOG.warn("Returning, interrupted : {}", String.valueOf(e));
break;
}
- LOG.info("Processing the event " + event.toString());
+ LOG.info("Processing the event {}", event);
if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
@@ -295,7 +296,7 @@ public void run() {
TaskAttemptId taId = event.getTaskAttemptID();
Future<?> future = futures.remove(taId);
if (future != null) {
- LOG.info("canceling the task attempt " + taId);
+ LOG.info("canceling the task attempt {}", taId);
future.cancel(true);
}
@@ -378,14 +379,12 @@ private void runTask(ContainerRemoteLaunchEvent launchEv,
// if umbilical itself barfs (in error-handler of runSubMap()),
// we're pretty much hosed, so do what YarnChild main() does
// (i.e., exit clumsily--but can never happen, so no worries!)
- LOG.error("oopsie... this can never happen: "
- + StringUtils.stringifyException(ioe));
+ LOG.error("oopsie... this can never happen: {}",
StringUtils.stringifyException(ioe));
ExitUtil.terminate(-1);
} finally {
// remove my future
if (futures.remove(attemptID) != null) {
- LOG.info("removed attempt " + attemptID +
- " from the futures to keep track of");
+ LOG.info("removed attempt {} from the futures to keep track of",
attemptID);
}
}
}
@@ -460,8 +459,8 @@ private void runSubtask(org.apache.hadoop.mapred.Task task,
// checking event queue is a tad wacky...but could enforce ordering
// (assuming no "lost events") at LocalMRAppMaster [CURRENT
BUG(?):
// doesn't send reduce event until maps all done]
- LOG.error("CONTAINER_REMOTE_LAUNCH contains a reduce task ("
- + attemptID + "), but not yet finished with maps");
+ LOG.error("CONTAINER_REMOTE_LAUNCH contains a reduce task ({}),"
+ + " but not yet finished with maps", attemptID);
throw new RuntimeException();
}
@@ -487,16 +486,15 @@ private void runSubtask(org.apache.hadoop.mapred.Task
task,
throw new RuntimeException();
} catch (Exception exception) {
- LOG.warn("Exception running local (uberized) 'child' : "
- + StringUtils.stringifyException(exception));
+ LOG.warn("Exception running local (uberized) 'child' : {}",
+ StringUtils.stringifyException(exception));
try {
if (task != null) {
// do cleanup for the task
task.taskCleanup(umbilical);
}
} catch (Exception e) {
- LOG.info("Exception cleaning up: "
- + StringUtils.stringifyException(e));
+ LOG.info("Exception cleaning up: {}",
StringUtils.stringifyException(e));
}
// Report back any failures, for diagnostic purposes
umbilical.reportDiagnosticInfo(classicAttemptID,
@@ -543,9 +541,8 @@ private void relocalize() {
deleted = false;
}
if (!deleted) {
- LOG.warn("Unable to delete unexpected local file/dir "
- + curLocalFiles[j].getName()
- + ": insufficient permissions?");
+ LOG.warn("Unable to delete unexpected local file/dir {}:
insufficient permissions?",
+ curLocalFiles[j].getName());
}
}
}
@@ -576,9 +573,8 @@ protected static MapOutputFile
renameMapOutputForReduce(JobConf conf,
Path mapOutIndex = subMapOutputFile.getOutputIndexFile();
Path reduceInIndex = new Path(reduceIn.toString() + ".index");
if (LOG.isDebugEnabled()) {
- LOG.debug("Renaming map output file for task attempt "
- + mapId.toString() + " from original location " + mapOut.toString()
- + " to destination " + reduceIn.toString());
+ LOG.debug("Renaming map output file for task attempt {} from original
location {}"
+ + " to destination {}", mapId, mapOut, reduceIn);
}
if (!localFs.mkdirs(reduceIn.getParent())) {
throw new IOException("Mkdirs failed to create "
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/TaskAttemptListenerImpl.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/TaskAttemptListenerImpl.java
index 5dffd735fda..f0bd4f4ff04 100644
---
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/TaskAttemptListenerImpl.java
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/TaskAttemptListenerImpl.java
@@ -34,9 +34,13 @@
import org.apache.hadoop.classification.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.ipc.ProtobufRpcEngine2;
import org.apache.hadoop.ipc.ProtocolSignature;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos;
+import org.apache.hadoop.mapred.protocolPB.TaskUmbilicalProtocolPB;
+import
org.apache.hadoop.mapred.protocolPB.TaskUmbilicalProtocolServerSideTranslatorPB;
import org.apache.hadoop.mapred.SortedRanges.Range;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.TypeConverter;
@@ -62,12 +66,13 @@
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.authorize.PolicyProvider;
import org.apache.hadoop.service.CompositeService;
+import org.apache.hadoop.thirdparty.protobuf.BlockingService;
import org.apache.hadoop.util.StringInterner;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
/**
- * This class is responsible for talking to the task umblical.
+ * This class is responsible for talking to the task umbilical.
* It also converts all the old data structures
* to yarn data structures.
*
@@ -82,32 +87,32 @@ public class TaskAttemptListenerImpl extends
CompositeService
private static final Logger LOG =
LoggerFactory.getLogger(TaskAttemptListenerImpl.class);
- private AppContext context;
+ private final AppContext context;
private Server server;
protected TaskHeartbeatHandler taskHeartbeatHandler;
- private RMHeartbeatHandler rmHeartbeatHandler;
+ private final RMHeartbeatHandler rmHeartbeatHandler;
private long commitWindowMs;
private InetSocketAddress address;
- private ConcurrentMap<WrappedJvmID, org.apache.hadoop.mapred.Task>
+ private final ConcurrentMap<WrappedJvmID, org.apache.hadoop.mapred.Task>
jvmIDToActiveAttemptMap
- = new ConcurrentHashMap<WrappedJvmID, org.apache.hadoop.mapred.Task>();
+ = new ConcurrentHashMap<>();
- private ConcurrentMap<TaskAttemptId,
+ private final ConcurrentMap<TaskAttemptId,
AtomicReference<TaskAttemptStatus>> attemptIdToStatus
= new ConcurrentHashMap<>();
/**
* A Map to keep track of the history of logging each task attempt.
*/
- private ConcurrentHashMap<TaskAttemptID, TaskProgressLogPair>
+ private final ConcurrentHashMap<TaskAttemptID, TaskProgressLogPair>
taskAttemptLogProgressStamps = new ConcurrentHashMap<>();
- private Set<WrappedJvmID> launchedJVMs = Collections
- .newSetFromMap(new ConcurrentHashMap<WrappedJvmID, Boolean>());
+ private final Set<WrappedJvmID> launchedJVMs = Collections
+ .newSetFromMap(new ConcurrentHashMap<>());
private JobTokenSecretManager jobTokenSecretManager = null;
- private AMPreemptionPolicy preemptionPolicy;
- private byte[] encryptedSpillKey;
+ private final AMPreemptionPolicy preemptionPolicy;
+ private final byte[] encryptedSpillKey;
public TaskAttemptListenerImpl(AppContext context,
JobTokenSecretManager jobTokenSecretManager,
@@ -155,11 +160,19 @@ protected void registerHeartbeatHandler(Configuration
conf) {
protected void startRpcServer() {
Configuration conf = getConfig();
try {
- server = new RPC.Builder(conf).setProtocol(TaskUmbilicalProtocol.class)
- .setInstance(this).setBindAddress("0.0.0.0")
+ RPC.setProtocolEngine(conf, TaskUmbilicalProtocolPB.class,
+ ProtobufRpcEngine2.class);
+ TaskUmbilicalProtocolServerSideTranslatorPB translator =
+ new TaskUmbilicalProtocolServerSideTranslatorPB(this);
+ BlockingService blockingService =
+ TaskUmbilicalProtocolProtos.TaskUmbilicalProtocolService
+ .newReflectiveBlockingService(translator);
+ server = new RPC.Builder(conf)
+ .setProtocol(TaskUmbilicalProtocolPB.class)
+ .setInstance(blockingService)
+ .setBindAddress("0.0.0.0")
.setPortRangeConfig(MRJobConfig.MR_AM_JOB_CLIENT_PORT_RANGE)
- .setNumHandlers(
- conf.getInt(MRJobConfig.MR_AM_TASK_LISTENER_THREAD_COUNT,
+
.setNumHandlers(conf.getInt(MRJobConfig.MR_AM_TASK_LISTENER_THREAD_COUNT,
MRJobConfig.DEFAULT_MR_AM_TASK_LISTENER_THREAD_COUNT))
.setVerbose(false).setSecretManager(jobTokenSecretManager).build();
@@ -186,8 +199,11 @@ void refreshServiceAcls(Configuration configuration,
@Override
protected void serviceStop() throws Exception {
- stopRpcServer();
- super.serviceStop();
+ try {
+ stopRpcServer();
+ } finally {
+ super.serviceStop();
+ }
}
protected void stopRpcServer() {
@@ -213,7 +229,7 @@ public InetSocketAddress getAddress() {
*/
@Override
public boolean canCommit(TaskAttemptID taskAttemptID) throws IOException {
- LOG.info("Commit go/no-go request from " + taskAttemptID.toString());
+ LOG.info("Commit go/no-go request from {}", taskAttemptID.toString());
// An attempt is asking if it can commit its output. This can be decided
// only by the task which is managing the multiple attempts. So redirect
the
// request there.
@@ -248,7 +264,7 @@ public boolean canCommit(TaskAttemptID taskAttemptID)
throws IOException {
@Override
public void commitPending(TaskAttemptID taskAttemptID, TaskStatus taskStatsu)
throws IOException, InterruptedException {
- LOG.info("Commit-pending state update from " + taskAttemptID.toString());
+ LOG.info("Commit-pending state update from {}", taskAttemptID.toString());
// An attempt is asking if it can commit its output. This can be decided
// only by the task which is managing the multiple attempts. So redirect
the
// request there.
@@ -265,7 +281,7 @@ public void commitPending(TaskAttemptID taskAttemptID,
TaskStatus taskStatsu)
@Override
public void preempted(TaskAttemptID taskAttemptID, TaskStatus taskStatus)
throws IOException, InterruptedException {
- LOG.info("Preempted state update from " + taskAttemptID.toString());
+ LOG.info("Preempted state update from {}", taskAttemptID.toString());
// An attempt is telling us that it got preempted.
org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID =
TypeConverter.toYarn(taskAttemptID);
@@ -280,7 +296,7 @@ public void preempted(TaskAttemptID taskAttemptID,
TaskStatus taskStatus)
@Override
public void done(TaskAttemptID taskAttemptID) throws IOException {
- LOG.info("Done acknowledgment from " + taskAttemptID.toString());
+ LOG.info("Done acknowledgment from {}", taskAttemptID.toString());
org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID =
TypeConverter.toYarn(taskAttemptID);
@@ -295,7 +311,7 @@ public void done(TaskAttemptID taskAttemptID) throws
IOException {
public void fatalError(TaskAttemptID taskAttemptID, String msg, boolean
fastFail)
throws IOException {
// This happens only in Child and in the Task.
- LOG.error("Task: " + taskAttemptID + " - exited : " + msg);
+ LOG.error("Task: {} - exited : {}", taskAttemptID, msg);
reportDiagnosticInfo(taskAttemptID, "Error: " + msg);
org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID =
@@ -312,8 +328,7 @@ public void fatalError(TaskAttemptID taskAttemptID, String
msg, boolean fastFail
public void fsError(TaskAttemptID taskAttemptID, String message)
throws IOException {
// This happens only in Child.
- LOG.error("Task: " + taskAttemptID + " - failed due to FSError: "
- + message);
+ LOG.error("Task: {} - failed due to FSError: {}", taskAttemptID, message);
reportDiagnosticInfo(taskAttemptID, "FSError: " + message);
org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID =
@@ -335,8 +350,8 @@ public void shuffleError(TaskAttemptID taskAttemptID,
String message) throws IOE
public MapTaskCompletionEventsUpdate getMapCompletionEvents(
JobID jobIdentifier, int startIndex, int maxEvents,
TaskAttemptID taskAttemptID) throws IOException {
- LOG.info("MapCompletionEvents request from " + taskAttemptID.toString()
- + ". startIndex " + startIndex + " maxEvents " + maxEvents);
+ LOG.info("MapCompletionEvents request from {}. startIndex {} maxEvents {}",
+ taskAttemptID.toString(), startIndex, maxEvents);
// TODO: shouldReset is never used. See TT. Ask for Removal.
boolean shouldReset = false;
@@ -355,8 +370,7 @@ public MapTaskCompletionEventsUpdate getMapCompletionEvents(
public void reportDiagnosticInfo(TaskAttemptID taskAttemptID, String
diagnosticInfo)
throws IOException {
diagnosticInfo = StringInterner.weakIntern(diagnosticInfo);
- LOG.info("Diagnostics report from " + taskAttemptID.toString() + ": "
- + diagnosticInfo);
+ LOG.info("Diagnostics report from {}: {}", taskAttemptID.toString(),
diagnosticInfo);
org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId attemptID =
TypeConverter.toYarn(taskAttemptID);
@@ -389,8 +403,7 @@ public AMFeedback statusUpdate(TaskAttemptID taskAttemptID,
// down gracefully or receiving a thread dump signal. Tolerate unknown
// tasks as long as they have unregistered recently.
if (!taskHeartbeatHandler.hasRecentlyUnregistered(yarnAttemptID)) {
- LOG.error("Status update was called with illegal TaskAttemptId: "
- + yarnAttemptID);
+ LOG.error("Status update was called with illegal TaskAttemptId: {}",
yarnAttemptID);
feedback.setTaskFound(false);
}
return feedback;
@@ -400,14 +413,14 @@ public AMFeedback statusUpdate(TaskAttemptID
taskAttemptID,
if (getConfig().getBoolean(MRJobConfig.TASK_PREEMPTION, false)
&& preemptionPolicy.isPreempted(yarnAttemptID)) {
feedback.setPreemption(true);
- LOG.info("Setting preemption bit for task: "+ yarnAttemptID
- + " of type " + yarnAttemptID.getTaskId().getTaskType());
+ LOG.info("Setting preemption bit for task: {} of type {}", yarnAttemptID,
+ yarnAttemptID.getTaskId().getTaskType());
}
if (taskStatus == null) {
//We are using statusUpdate only as a simple ping
if (LOG.isDebugEnabled()) {
- LOG.debug("Ping from " + taskAttemptID.toString());
+ LOG.debug("Ping from {}", taskAttemptID.toString());
}
// Consider ping from the tasks for liveliness check
if
(getConfig().getBoolean(MRJobConfig.MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK,
@@ -459,9 +472,9 @@ public AMFeedback statusUpdate(TaskAttemptID taskAttemptID,
//set the fetch failures
if (taskStatus.getFetchFailedMaps() != null
- && taskStatus.getFetchFailedMaps().size() > 0) {
- taskAttemptStatus.fetchFailedMaps =
- new
ArrayList<org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId>();
+ && !taskStatus.getFetchFailedMaps().isEmpty()) {
+ taskAttemptStatus.fetchFailedMaps =
+ new ArrayList<>();
for (TaskAttemptID failedMapId : taskStatus.getFetchFailedMaps()) {
taskAttemptStatus.fetchFailedMaps.add(
TypeConverter.toYarn(failedMapId));
@@ -505,7 +518,7 @@ public JvmTask getTask(JvmContext context) throws
IOException {
// A rough imitation of code from TaskTracker.
JVMId jvmId = context.jvmId;
- LOG.info("JVM with ID : " + jvmId + " asked for a task");
+ LOG.info("JVM with ID : {} asked for a task", jvmId);
JvmTask jvmTask = null;
// TODO: Is it an authorized container to get a task? Otherwise return
null.
@@ -519,13 +532,13 @@ public JvmTask getTask(JvmContext context) throws
IOException {
// Try to look up the task. We remove it directly as we don't give
// multiple tasks to a JVM
if (!jvmIDToActiveAttemptMap.containsKey(wJvmID)) {
- LOG.info("JVM with ID: " + jvmId + " is invalid and will be killed.");
+ LOG.info("JVM with ID: {} is invalid and will be killed.", jvmId);
jvmTask = TASK_FOR_INVALID_JVM;
} else {
if (!launchedJVMs.contains(wJvmID)) {
jvmTask = null;
- LOG.info("JVM with ID: " + jvmId
- + " asking for task before AM launch registered. Given null task");
+ LOG.info("JVM with ID: {} asking for task before AM launch registered.
Given null task",
+ jvmId);
} else {
// remove the task as it is no more needed and free up the memory.
// Also we have already told the JVM to process a task, so it is no
@@ -533,7 +546,7 @@ public JvmTask getTask(JvmContext context) throws
IOException {
org.apache.hadoop.mapred.Task task =
jvmIDToActiveAttemptMap.remove(wJvmID);
launchedJVMs.remove(wJvmID);
- LOG.info("JVM with ID: " + jvmId + " given task: " + task.getTaskID());
+ LOG.info("JVM with ID: {} given task: {}", jvmId, task.getTaskID());
task.setEncryptedSpillKey(encryptedSpillKey);
jvmTask = new JvmTask(task, false);
}
@@ -594,7 +607,7 @@ public ProtocolSignature getProtocolSignature(String
protocol,
protocol, clientVersion, clientMethodsHash);
}
- // task checkpoint bookeeping
+ // task checkpoint bookkeeping
@Override
public TaskCheckpointID getCheckpointID(TaskID taskId) {
TaskId tid = TypeConverter.toYarn(taskId);
@@ -634,8 +647,8 @@ private void coalesceStatusUpdate(TaskAttemptId
yarnAttemptID,
// it processes the update, or by another IPC server handler
done = lastStatusRef.compareAndSet(lastStatus, taskAttemptStatus);
if (!done) {
- LOG.info("TaskAttempt " + yarnAttemptID +
- ": lastStatusRef changed by another thread, retrying...");
+ LOG.info("TaskAttempt {}: lastStatusRef changed by another thread,
retrying...",
+ yarnAttemptID);
// let's revert taskAttemptStatus.fetchFailedMaps
taskAttemptStatus.fetchFailedMaps = fetchFailedMaps;
}
@@ -687,12 +700,10 @@ private void resetLog(final boolean doLog,
if (doLog) {
prevProgress = processedProgress;
logTimeStamp = timestamp;
- LOG.info("Progress of TaskAttempt " + taskAttemptID + " is : "
- + progress);
+ LOG.info("Progress of TaskAttempt {} is : {}", taskAttemptID,
progress);
} else {
if (LOG.isDebugEnabled()) {
- LOG.debug("Progress of TaskAttempt " + taskAttemptID + " is : "
- + progress);
+ LOG.debug("Progress of TaskAttempt {} is : {}", taskAttemptID,
progress);
}
}
}
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/YarnChild.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/YarnChild.java
index bbf527ebff5..ab5e241262b 100644
---
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/YarnChild.java
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/YarnChild.java
@@ -19,7 +19,9 @@
package org.apache.hadoop.mapred;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.hadoop.io.IOUtils.closeStream;
+import java.io.Closeable;
import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
@@ -35,7 +37,10 @@
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.ipc.CallerContext;
+import org.apache.hadoop.ipc.ProtobufRpcEngine2;
import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.mapred.protocolPB.TaskUmbilicalProtocolPB;
+import org.apache.hadoop.mapred.protocolPB.TaskUmbilicalProtocolPBClientImpl;
import org.apache.hadoop.mapreduce.MRConfig;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.TaskType;
@@ -116,9 +121,14 @@ public static void main(String[] args) throws Throwable {
taskOwner.doAs(new PrivilegedExceptionAction<TaskUmbilicalProtocol>() {
@Override
public TaskUmbilicalProtocol run() throws Exception {
- return (TaskUmbilicalProtocol)RPC.getProxy(TaskUmbilicalProtocol.class,
- TaskUmbilicalProtocol.versionID, address, job);
- }
+ RPC.setProtocolEngine(job, TaskUmbilicalProtocolPB.class,
+ ProtobufRpcEngine2.class);
+ TaskUmbilicalProtocolPB proxy = RPC.getProxy(
+ TaskUmbilicalProtocolPB.class,
+ RPC.getProtocolVersion(TaskUmbilicalProtocolPB.class),
+ address, job);
+ return new TaskUmbilicalProtocolPBClientImpl(proxy);
+ }
});
// report non-pid to application master
@@ -185,8 +195,7 @@ public Object run() throws Exception {
umbilical.fsError(taskid, e.getMessage());
}
} catch (Exception exception) {
- LOG.warn("Exception running child : "
- + StringUtils.stringifyException(exception));
+ LOG.warn("Exception running child : {}",
StringUtils.stringifyException(exception));
try {
if (task != null) {
// do cleanup for the task
@@ -194,17 +203,14 @@ public Object run() throws Exception {
task.taskCleanup(umbilical);
} else {
final Task taskFinal = task;
- childUGI.doAs(new PrivilegedExceptionAction<Object>() {
- @Override
- public Object run() throws Exception {
- taskFinal.taskCleanup(umbilical);
- return null;
- }
+ childUGI.doAs((PrivilegedExceptionAction<Object>) () -> {
+ taskFinal.taskCleanup(umbilical);
+ return null;
});
}
}
} catch (Exception e) {
- LOG.info("Exception cleaning up: " +
StringUtils.stringifyException(e));
+ LOG.info("Exception cleaning up: {}",
StringUtils.stringifyException(e));
}
// Report back any failures, for diagnostic purposes
if (taskid != null) {
@@ -213,8 +219,7 @@ public Object run() throws Exception {
}
}
} catch (Throwable throwable) {
- LOG.error("Error running child : "
- + StringUtils.stringifyException(throwable));
+ LOG.error("Error running child : {}",
StringUtils.stringifyException(throwable));
if (taskid != null) {
if (!ShutdownHookManager.get().isShutdownInProgress()) {
Throwable tCause = throwable.getCause();
@@ -225,7 +230,11 @@ public Object run() throws Exception {
}
}
} finally {
- RPC.stopProxy(umbilical);
+ if (umbilical instanceof Closeable closeable) {
+ closeStream(closeable);
+ } else {
+ RPC.stopProxy(umbilical);
+ }
DefaultMetricsSystem.shutdown();
TaskLog.syncLogsShutdown(logSyncer);
}
@@ -278,7 +287,7 @@ private static void configureLocalDirs(Task task, JobConf
job) throws IOExceptio
String[] localSysDirs = StringUtils.getTrimmedStrings(
System.getenv(Environment.LOCAL_DIRS.name()));
job.setStrings(MRConfig.LOCAL_DIR, localSysDirs);
- LOG.info(MRConfig.LOCAL_DIR + " for child: " +
job.get(MRConfig.LOCAL_DIR));
+ LOG.info(MRConfig.LOCAL_DIR + " for child: {}",
job.get(MRConfig.LOCAL_DIR));
LocalDirAllocator lDirAlloc = new LocalDirAllocator(MRConfig.LOCAL_DIR);
Path workDir = null;
// First, try to find the JOB_LOCAL_DIR on this host.
@@ -318,7 +327,7 @@ private static void configureTask(JobConf job, Task task,
ApplicationAttemptId appAttemptId = ContainerId.fromString(
System.getenv(Environment.CONTAINER_ID.name()))
.getApplicationAttemptId();
- LOG.debug("APPLICATION_ATTEMPT_ID: " + appAttemptId);
+ LOG.debug("APPLICATION_ATTEMPT_ID: {}", appAttemptId);
// Set it in conf, so as to be able to be used the the OutputCommitter.
job.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID,
appAttemptId.getAttemptId());
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolPB.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolPB.java
new file mode 100644
index 00000000000..c3814cb9b01
--- /dev/null
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolPB.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mapred.protocolPB;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.ipc.ProtocolInfo;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.TaskUmbilicalProtocolService;
+import org.apache.hadoop.mapreduce.security.token.JobTokenSelector;
+import org.apache.hadoop.security.token.TokenInfo;
+
+/**
+ * Protocol buffer based RPC interface for {@link
org.apache.hadoop.mapred.TaskUmbilicalProtocol}.
+ * Protocol version 1 corresponds to the TaskUmbilicalProtocol.versionID used
in
+ * the Writable-based RPC implementation.
+ */
[email protected]
+@ProtocolInfo(
+ protocolName = "org.apache.hadoop.mapred.TaskUmbilicalProtocol",
+ protocolVersion = 1)
+@TokenInfo(JobTokenSelector.class)
+public interface TaskUmbilicalProtocolPB
+ extends TaskUmbilicalProtocolService.BlockingInterface {
+}
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolPBClientImpl.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolPBClientImpl.java
new file mode 100644
index 00000000000..bc9fb0c1be0
--- /dev/null
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolPBClientImpl.java
@@ -0,0 +1,337 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mapred.protocolPB;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.ipc.ProtocolSignature;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.mapred.AMFeedback;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JvmContext;
+import org.apache.hadoop.mapred.JvmTask;
+import org.apache.hadoop.mapred.MapTaskCompletionEventsUpdate;
+import org.apache.hadoop.mapred.SortedRanges;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskID;
+import org.apache.hadoop.mapred.TaskStatus;
+import org.apache.hadoop.mapred.TaskUmbilicalProtocol;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.CanCommitRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.CommitPendingRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.DoneRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.FatalErrorRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.FsErrorRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetCheckpointIDRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetCheckpointIDResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetMapCompletionEventsRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetMapCompletionEventsResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetTaskRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetTaskResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.PreemptedRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.ReportDiagnosticInfoRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.ReportNextRecordRangeRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.SetCheckpointIDRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.ShuffleErrorRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.StatusUpdateRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.StatusUpdateResponseProto;
+import org.apache.hadoop.mapreduce.checkpoint.TaskCheckpointID;
+
+import static org.apache.hadoop.ipc.RPC.stopProxy;
+import static
org.apache.hadoop.mapred.protocolPB.TaskUmbilicalProtocolUtils.service;
+
+/**
+ * Client-side translator to translate calls from {@link TaskUmbilicalProtocol}
+ * to the RPC server implementing {@link TaskUmbilicalProtocolPB}.
+ */
[email protected]
+public class TaskUmbilicalProtocolPBClientImpl
+ implements TaskUmbilicalProtocol, Closeable {
+
+ private final TaskUmbilicalProtocolPB proxy;
+
+ public TaskUmbilicalProtocolPBClientImpl(TaskUmbilicalProtocolPB proxy) {
+ this.proxy = proxy;
+ }
+
+ @Override
+ public void close() throws IOException {
+ stopProxy(proxy);
+ }
+
+ @Override
+ public long getProtocolVersion(String protocol, long clientVersion)
+ throws IOException {
+ return RPC.getProtocolVersion(TaskUmbilicalProtocolPB.class);
+ }
+
+ @Override
+ public ProtocolSignature getProtocolSignature(
+ String protocol, long clientVersion, int clientMethodsHash)
+ throws IOException {
+ return ProtocolSignature.getProtocolSignature(this,
+ protocol, clientVersion, clientMethodsHash);
+ }
+
+ @Override
+ public JvmTask getTask(JvmContext context) throws IOException {
+ return service(() -> {
+ GetTaskRequestProto.Builder builder = GetTaskRequestProto.newBuilder();
+ if (context != null) {
+ builder.setJvmContext(
+ TaskUmbilicalProtocolUtils.serialize(context));
+ }
+ GetTaskResponseProto response =
+ proxy.getTask(null, builder.build());
+ if (!response.hasJvmTask()) {
+ return null;
+ }
+ return TaskUmbilicalProtocolUtils.deserialize(
+ new JvmTask(), response.getJvmTask());
+ });
+ }
+
+ @Override
+ public AMFeedback statusUpdate(TaskAttemptID taskId, TaskStatus taskStatus)
+ throws IOException, InterruptedException {
+ return service(() -> {
+ StatusUpdateRequestProto.Builder builder =
+ StatusUpdateRequestProto.newBuilder();
+ if (taskId != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskId));
+ }
+ if (taskStatus != null) {
+ builder.setTaskStatus(
+ TaskUmbilicalProtocolUtils.serializeTaskStatus(
+ taskStatus));
+ }
+ StatusUpdateResponseProto response =
+ proxy.statusUpdate(null, builder.build());
+ AMFeedback feedback = new AMFeedback();
+ feedback.setTaskFound(response.getTaskFound());
+ feedback.setPreemption(response.getPreemption());
+ return feedback;
+ });
+ }
+
+ @Override
+ public void reportDiagnosticInfo(TaskAttemptID taskid, String trace)
+ throws IOException {
+ service(() -> {
+ ReportDiagnosticInfoRequestProto.Builder builder =
+ ReportDiagnosticInfoRequestProto.newBuilder();
+ if (taskid != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskid));
+ }
+ builder.setTrace(trace != null ? trace : "");
+ proxy.reportDiagnosticInfo(null, builder.build());
+ });
+ }
+
+ @Override
+ public void reportNextRecordRange(TaskAttemptID taskid,
+ SortedRanges.Range range) throws IOException {
+ service(() -> {
+ ReportNextRecordRangeRequestProto.Builder builder =
+ ReportNextRecordRangeRequestProto.newBuilder();
+ if (taskid != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskid));
+ }
+ if (range != null) {
+ builder.setRange(
+ TaskUmbilicalProtocolUtils.serialize(range));
+ }
+ proxy.reportNextRecordRange(null, builder.build());
+ });
+ }
+
+ @Override
+ public void done(TaskAttemptID taskid) throws IOException {
+ service(() -> {
+ DoneRequestProto.Builder builder = DoneRequestProto.newBuilder();
+ if (taskid != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskid));
+ }
+ proxy.done(null, builder.build());
+ });
+ }
+
+ @Override
+ public void commitPending(TaskAttemptID taskId, TaskStatus taskStatus)
+ throws IOException, InterruptedException {
+ service(() -> {
+ CommitPendingRequestProto.Builder builder =
+ CommitPendingRequestProto.newBuilder();
+ if (taskId != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskId));
+ }
+ if (taskStatus != null) {
+ builder.setTaskStatus(
+ TaskUmbilicalProtocolUtils.serializeTaskStatus(
+ taskStatus));
+ }
+ proxy.commitPending(null, builder.build());
+ });
+ }
+
+ @Override
+ public boolean canCommit(TaskAttemptID taskid) throws IOException {
+ return service(() -> {
+ CanCommitRequestProto.Builder builder =
+ CanCommitRequestProto.newBuilder();
+ if (taskid != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskid));
+ }
+ return proxy.canCommit(null, builder.build()).getCanCommit();
+ });
+ }
+
+ @Override
+ public void shuffleError(TaskAttemptID taskId, String message)
+ throws IOException {
+ service(() -> {
+ ShuffleErrorRequestProto.Builder builder =
+ ShuffleErrorRequestProto.newBuilder();
+ if (taskId != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskId));
+ }
+ builder.setMessage(message != null ? message : "");
+ proxy.shuffleError(null, builder.build());
+ });
+ }
+
+ @Override
+ public void fsError(TaskAttemptID taskId, String message) throws IOException
{
+ service(() -> {
+ FsErrorRequestProto.Builder builder = FsErrorRequestProto.newBuilder();
+ if (taskId != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskId));
+ }
+ builder.setMessage(message != null ? message : "");
+ proxy.fsError(null, builder.build());
+ });
+ }
+
+ @Override
+ public void fatalError(TaskAttemptID taskId, String msg, boolean fastFail)
+ throws IOException {
+ service(() -> {
+ FatalErrorRequestProto.Builder builder =
+ FatalErrorRequestProto.newBuilder();
+ if (taskId != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskId));
+ }
+ builder.setMessage(msg != null ? msg : "");
+ builder.setFastFail(fastFail);
+ proxy.fatalError(null, builder.build());
+ });
+ }
+
+ @Override
+ public MapTaskCompletionEventsUpdate getMapCompletionEvents(JobID jobId,
+ int fromIndex, int maxLocs, TaskAttemptID id) throws IOException {
+ return service(() -> {
+ GetMapCompletionEventsRequestProto.Builder builder =
+ GetMapCompletionEventsRequestProto.newBuilder();
+ if (jobId != null) {
+ builder.setJobId(
+ TaskUmbilicalProtocolUtils.serialize(jobId));
+ }
+ builder.setFromIndex(fromIndex);
+ builder.setMaxLocs(maxLocs);
+ if (id != null) {
+ builder.setTaskAttemptId(
+ TaskUmbilicalProtocolUtils.serialize(id));
+ }
+ GetMapCompletionEventsResponseProto response =
+ proxy.getMapCompletionEvents(null, builder.build());
+ if (!response.hasEventsUpdate()) {
+ return null;
+ }
+ return TaskUmbilicalProtocolUtils.deserialize(
+ new MapTaskCompletionEventsUpdate(), response.getEventsUpdate());
+ });
+ }
+
+ @Override
+ public void preempted(TaskAttemptID taskId, TaskStatus taskStatus)
+ throws IOException, InterruptedException {
+ service(() -> {
+ PreemptedRequestProto.Builder builder =
PreemptedRequestProto.newBuilder();
+ if (taskId != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskId));
+ }
+ if (taskStatus != null) {
+ builder.setTaskStatus(
+ TaskUmbilicalProtocolUtils.serializeTaskStatus(
+ taskStatus));
+ }
+ proxy.preempted(null, builder.build());
+ });
+ }
+
+ @Override
+ public TaskCheckpointID getCheckpointID(TaskID taskId) {
+ return service(() -> {
+ GetCheckpointIDRequestProto.Builder builder =
+ GetCheckpointIDRequestProto.newBuilder();
+ if (taskId != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskId));
+ }
+ GetCheckpointIDResponseProto response =
+ proxy.getCheckpointID(null, builder.build());
+ if (!response.hasCheckpointId()) {
+ return null;
+ }
+ return TaskUmbilicalProtocolUtils.deserialize(
+ new TaskCheckpointID(), response.getCheckpointId());
+ });
+ }
+
+ @Override
+ public void setCheckpointID(TaskID taskId, TaskCheckpointID checkpointId) {
+ service(() -> {
+ SetCheckpointIDRequestProto.Builder builder =
+ SetCheckpointIDRequestProto.newBuilder();
+ if (taskId != null) {
+ builder.setTaskId(
+ TaskUmbilicalProtocolUtils.serialize(taskId));
+ }
+ if (checkpointId != null) {
+ builder.setCheckpointId(
+ TaskUmbilicalProtocolUtils.serialize(checkpointId));
+ }
+ proxy.setCheckpointID(null, builder.build());
+ });
+ }
+
+
+}
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolServerSideTranslatorPB.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolServerSideTranslatorPB.java
new file mode 100644
index 00000000000..f31a4aa1eb5
--- /dev/null
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolServerSideTranslatorPB.java
@@ -0,0 +1,322 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mapred.protocolPB;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.mapred.AMFeedback;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JvmContext;
+import org.apache.hadoop.mapred.JvmTask;
+import org.apache.hadoop.mapred.MapTaskCompletionEventsUpdate;
+import org.apache.hadoop.mapred.SortedRanges;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskID;
+import org.apache.hadoop.mapred.TaskStatus;
+import org.apache.hadoop.mapred.TaskUmbilicalProtocol;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.CanCommitRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.CanCommitResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.CommitPendingRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.CommitPendingResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.DoneRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.DoneResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.FatalErrorRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.FatalErrorResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.FsErrorRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.FsErrorResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetCheckpointIDRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetCheckpointIDResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetMapCompletionEventsRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetMapCompletionEventsResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetTaskRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.GetTaskResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.PreemptedRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.PreemptedResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.ReportDiagnosticInfoRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.ReportDiagnosticInfoResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.ReportNextRecordRangeRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.ReportNextRecordRangeResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.SetCheckpointIDRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.SetCheckpointIDResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.ShuffleErrorRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.ShuffleErrorResponseProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.StatusUpdateRequestProto;
+import
org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos.StatusUpdateResponseProto;
+import org.apache.hadoop.mapreduce.checkpoint.TaskCheckpointID;
+import org.apache.hadoop.thirdparty.protobuf.RpcController;
+import org.apache.hadoop.thirdparty.protobuf.ServiceException;
+
+import static
org.apache.hadoop.mapred.protocolPB.TaskUmbilicalProtocolUtils.translate;
+
+/**
+ * Server-side translator to translate the requests received on
+ * {@link TaskUmbilicalProtocolPB} to the calls in {@link
TaskUmbilicalProtocol}.
+ */
[email protected]
+public class TaskUmbilicalProtocolServerSideTranslatorPB
+ implements TaskUmbilicalProtocolPB {
+
+ private static final ReportDiagnosticInfoResponseProto
VOID_REPORT_DIAG_RESPONSE =
+ ReportDiagnosticInfoResponseProto.newBuilder().build();
+
+ private static final ReportNextRecordRangeResponseProto
VOID_REPORT_RANGE_RESPONSE =
+ ReportNextRecordRangeResponseProto.newBuilder().build();
+
+ private static final DoneResponseProto VOID_DONE_RESPONSE =
+ DoneResponseProto.newBuilder().build();
+
+ private static final CommitPendingResponseProto VOID_COMMIT_PENDING_RESPONSE
=
+ CommitPendingResponseProto.newBuilder().build();
+
+ private static final ShuffleErrorResponseProto VOID_SHUFFLE_ERROR_RESPONSE =
+ ShuffleErrorResponseProto.newBuilder().build();
+
+ private static final FsErrorResponseProto VOID_FS_ERROR_RESPONSE =
+ FsErrorResponseProto.newBuilder().build();
+
+ private static final FatalErrorResponseProto VOID_FATAL_ERROR_RESPONSE =
+ FatalErrorResponseProto.newBuilder().build();
+
+ private static final PreemptedResponseProto VOID_PREEMPTED_RESPONSE =
+ PreemptedResponseProto.newBuilder().build();
+
+ private static final SetCheckpointIDResponseProto
VOID_SET_CHECKPOINT_RESPONSE =
+ SetCheckpointIDResponseProto.newBuilder().build();
+
+
+ private final TaskUmbilicalProtocol impl;
+
+ public TaskUmbilicalProtocolServerSideTranslatorPB(TaskUmbilicalProtocol
impl) {
+ this.impl = impl;
+ }
+
+ @Override
+ public GetTaskResponseProto getTask(RpcController controller,
+ GetTaskRequestProto request) throws ServiceException {
+ return translate(() -> {
+ JvmContext context = request.hasJvmContext()
+ ? TaskUmbilicalProtocolUtils.deserialize(new JvmContext(),
request.getJvmContext())
+ : null;
+ JvmTask task = impl.getTask(context);
+ GetTaskResponseProto.Builder builder = GetTaskResponseProto.newBuilder();
+ if (task != null) {
+ builder.setJvmTask(TaskUmbilicalProtocolUtils.serialize(task));
+ }
+ return builder.build();
+ });
+ }
+
+ @Override
+ public StatusUpdateResponseProto statusUpdate(RpcController controller,
+ StatusUpdateRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskAttemptID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskAttemptID(),
request.getTaskId())
+ : null;
+ TaskStatus taskStatus = null;
+ if (request.hasTaskStatus()) {
+ taskStatus =
TaskUmbilicalProtocolUtils.deserializeTaskStatus(request.getTaskStatus());
+ }
+ AMFeedback feedback = impl.statusUpdate(taskId, taskStatus);
+ return StatusUpdateResponseProto.newBuilder()
+ .setTaskFound(feedback.getTaskFound())
+ .setPreemption(feedback.getPreemption())
+ .build();
+ });
+ }
+
+ @Override
+ public ReportDiagnosticInfoResponseProto reportDiagnosticInfo(
+ RpcController controller,
+ ReportDiagnosticInfoRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskAttemptID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskAttemptID(),
request.getTaskId())
+ : null;
+ impl.reportDiagnosticInfo(taskId, request.getTrace());
+ return VOID_REPORT_DIAG_RESPONSE;
+ });
+ }
+
+ @Override
+ public ReportNextRecordRangeResponseProto reportNextRecordRange(
+ RpcController controller,
+ ReportNextRecordRangeRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskAttemptID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskAttemptID(),
request.getTaskId())
+ : null;
+ SortedRanges.Range range = request.hasRange()
+ ? TaskUmbilicalProtocolUtils.deserialize(new SortedRanges.Range(),
request.getRange())
+ : null;
+ impl.reportNextRecordRange(taskId, range);
+ return VOID_REPORT_RANGE_RESPONSE;
+ });
+ }
+
+ @Override
+ public DoneResponseProto done(RpcController controller,
+ DoneRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskAttemptID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskAttemptID(),
request.getTaskId())
+ : null;
+ impl.done(taskId);
+ return VOID_DONE_RESPONSE;
+ });
+ }
+
+ @Override
+ public CommitPendingResponseProto commitPending(RpcController controller,
+ CommitPendingRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskAttemptID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskAttemptID(),
request.getTaskId())
+ : null;
+ TaskStatus taskStatus = null;
+ if (request.hasTaskStatus()) {
+ taskStatus =
TaskUmbilicalProtocolUtils.deserializeTaskStatus(request.getTaskStatus());
+ }
+ impl.commitPending(taskId, taskStatus);
+ return VOID_COMMIT_PENDING_RESPONSE;
+ });
+ }
+
+ @Override
+ public CanCommitResponseProto canCommit(RpcController controller,
+ CanCommitRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskAttemptID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskAttemptID(),
request.getTaskId())
+ : null;
+ boolean canCommit = impl.canCommit(taskId);
+ return CanCommitResponseProto.newBuilder()
+ .setCanCommit(canCommit)
+ .build();
+ });
+ }
+
+ @Override
+ public ShuffleErrorResponseProto shuffleError(RpcController controller,
+ ShuffleErrorRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskAttemptID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskAttemptID(),
request.getTaskId())
+ : null;
+ impl.shuffleError(taskId, request.getMessage());
+ return VOID_SHUFFLE_ERROR_RESPONSE;
+ });
+ }
+
+ @Override
+ public FsErrorResponseProto fsError(RpcController controller,
+ FsErrorRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskAttemptID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskAttemptID(),
request.getTaskId())
+ : null;
+ impl.fsError(taskId, request.getMessage());
+ return VOID_FS_ERROR_RESPONSE;
+ });
+ }
+
+ @Override
+ public FatalErrorResponseProto fatalError(RpcController controller,
+ FatalErrorRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskAttemptID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskAttemptID(),
request.getTaskId())
+ : null;
+ impl.fatalError(taskId, request.getMessage(), request.getFastFail());
+ return VOID_FATAL_ERROR_RESPONSE;
+ });
+ }
+
+ @Override
+ public GetMapCompletionEventsResponseProto getMapCompletionEvents(
+ RpcController controller,
+ GetMapCompletionEventsRequestProto request) throws ServiceException {
+ return translate(() -> {
+ JobID jobId = request.hasJobId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new JobID(),
request.getJobId())
+ : null;
+ TaskAttemptID taskAttemptId = request.hasTaskAttemptId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskAttemptID(),
request.getTaskAttemptId())
+ : null;
+ MapTaskCompletionEventsUpdate update = impl.getMapCompletionEvents(
+ jobId, request.getFromIndex(), request.getMaxLocs(), taskAttemptId);
+ GetMapCompletionEventsResponseProto.Builder builder =
+ GetMapCompletionEventsResponseProto.newBuilder();
+ if (update != null) {
+ builder.setEventsUpdate(TaskUmbilicalProtocolUtils.serialize(update));
+ }
+ return builder.build();
+ });
+ }
+
+ @Override
+ public PreemptedResponseProto preempted(RpcController controller,
+ PreemptedRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskAttemptID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskAttemptID(),
request.getTaskId())
+ : null;
+ TaskStatus taskStatus = null;
+ if (request.hasTaskStatus()) {
+ taskStatus =
TaskUmbilicalProtocolUtils.deserializeTaskStatus(request.getTaskStatus());
+ }
+ impl.preempted(taskId, taskStatus);
+ return VOID_PREEMPTED_RESPONSE;
+ });
+ }
+
+ @Override
+ public GetCheckpointIDResponseProto getCheckpointID(RpcController controller,
+ GetCheckpointIDRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskID(),
request.getTaskId())
+ : null;
+ TaskCheckpointID checkpointID = impl.getCheckpointID(taskId);
+ GetCheckpointIDResponseProto.Builder builder =
+ GetCheckpointIDResponseProto.newBuilder();
+ if (checkpointID != null) {
+
builder.setCheckpointId(TaskUmbilicalProtocolUtils.serialize(checkpointID));
+ }
+ return builder.build();
+ });
+ }
+
+ @Override
+ public SetCheckpointIDResponseProto setCheckpointID(RpcController controller,
+ SetCheckpointIDRequestProto request) throws ServiceException {
+ return translate(() -> {
+ TaskID taskId = request.hasTaskId()
+ ? TaskUmbilicalProtocolUtils.deserialize(new TaskID(),
request.getTaskId())
+ : null;
+ TaskCheckpointID checkpointID = null;
+ if (request.hasCheckpointId()) {
+ checkpointID = TaskUmbilicalProtocolUtils.deserialize(new
TaskCheckpointID(),
+ request.getCheckpointId());
+ }
+ impl.setCheckpointID(taskId, checkpointID);
+ return VOID_SET_CHECKPOINT_RESPONSE;
+ });
+ }
+
+}
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolUtils.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolUtils.java
new file mode 100644
index 00000000000..9ed7d5d42ef
--- /dev/null
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/TaskUmbilicalProtocolUtils.java
@@ -0,0 +1,190 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mapred.protocolPB;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapred.TaskStatus;
+import org.apache.hadoop.thirdparty.protobuf.ByteString;
+import org.apache.hadoop.thirdparty.protobuf.ServiceException;
+
+import static
org.apache.hadoop.ipc.internal.ShadedProtobufHelper.getRemoteException;
+
+/**
+ * Protocol helpers.
+ * <p>The lambda expressions and invokers are very similar to that of
ShadedProtobufHelper.ipc(),
+ * but with broader exception catching.
+ */
+final class TaskUmbilicalProtocolUtils {
+
+ private TaskUmbilicalProtocolUtils() {
+ }
+
+ /*
+ Serialization helpers
+ */
+
+ /**
+ * Serialize a Writable into a protobuf ByteString.
+ * @param writable source
+ * @return data written to bytes.
+ * @throws IOException write failure.
+ */
+ static ByteString serialize(Writable writable)
+ throws IOException {
+ DataOutputBuffer dob = new DataOutputBuffer();
+ writable.write(dob);
+ return ByteString.copyFrom(dob.getData(), 0, dob.getLength());
+ }
+
+ /**
+ * Deserialize specific writable from bytes in a protobuf string.
+ * @param writable the writable instance which will be filled with the data
+ * @param bytes bytes to unmarshall.
+ * @param <T> type of the writable.
+ * @return the writable.
+ * @throws IOException read failure.
+ */
+ static <T extends Writable> T deserialize(
+ T writable, ByteString bytes) throws IOException {
+ DataInputBuffer dib = new DataInputBuffer();
+ byte[] b = bytes.toByteArray();
+ dib.reset(b, b.length);
+ writable.readFields(dib);
+ return writable;
+ }
+
+ /**
+ * Serialize a task status.
+ * @param status task to serialize
+ * @return the serialized status, including a flag to indicate status type.
+ * @throws IOException write failre.
+ */
+ static ByteString serializeTaskStatus(TaskStatus status) throws IOException {
+ DataOutputBuffer dob = new DataOutputBuffer();
+ TaskStatus.writeTaskStatusForPB(dob, status);
+ return ByteString.copyFrom(dob.getData(), 0, dob.getLength());
+ }
+
+ /**
+ * Deserialize a byte string to a matching TaskStatus implementation.
+ * @param bytes source bytes
+ * @return task status
+ * @throws IOException read failure.
+ */
+ static TaskStatus deserializeTaskStatus(ByteString bytes) throws IOException
{
+ DataInputBuffer dib = new DataInputBuffer();
+ byte[] b = bytes.toByteArray();
+ dib.reset(b, b.length);
+ return TaskStatus.readTaskStatusFromPB(dib);
+ }
+
+ /*
+ Service invocation with exception translation.
+ */
+
+ /**
+ * Service invocation.
+ * @param <Result> type of result.
+ */
+ @FunctionalInterface
+ interface ServiceCallable<Result> {
+
+ Result apply() throws IOException, ServiceException;
+ }
+
+ /**
+ * Service invocation with void response.
+ */
+ @FunctionalInterface
+ interface ServiceVoidCallable {
+
+ void apply() throws IOException, ServiceException;
+ }
+
+ /**
+ * Invoke a service.
+ * @param callable callable
+ * @param <Result> type of result
+ * @return result
+ * @throws UncheckedIOException wrapped IOE.
+ */
+ static <Result> Result service(ServiceCallable<Result> callable) {
+ try {
+ return callable.apply();
+ } catch (ServiceException e) {
+ throw new UncheckedIOException(getRemoteException(e));
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ /**
+ * Invoke a service with no result.
+ * @param callable callable
+ * @throws UncheckedIOException wrapped IOE.
+ */
+ static void service(ServiceVoidCallable callable) {
+ service(() -> {
+ callable.apply();
+ return null;
+ });
+ }
+
+
+ /**
+ * Client side lambda expression.
+ * @param <Result> type of result.
+ */
+ @FunctionalInterface
+ interface ClientCallable<Result> {
+
+ Result apply() throws IOException, InterruptedException;
+ }
+
+ /**
+ * Client side lambda expression with void result.
+ */
+ @FunctionalInterface
+ interface ClientVoidCallable {
+
+ void apply() throws IOException, InterruptedException;
+ }
+
+
+ static <Result> Result translate(ClientCallable<Result> callable) throws
ServiceException {
+ try {
+ return callable.apply();
+ } catch (IOException | InterruptedException e) {
+ throw new ServiceException(e);
+ }
+ }
+
+ static void translate(ClientVoidCallable callable) throws ServiceException {
+ translate(() -> {
+ callable.apply();
+ return null;
+ });
+ }
+
+}
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JvmContext.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/package-info.java
similarity index 50%
copy from
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JvmContext.java
copy to
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/package-info.java
index 88eeefc8e76..5904dd28dcd 100644
---
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JvmContext.java
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/protocolPB/package-info.java
@@ -1,4 +1,4 @@
-/**
+/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -16,42 +16,17 @@
* limitations under the License.
*/
-package org.apache.hadoop.mapred;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.Writable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+/**
+ * This package contains the protobuf implementation of the MR
+ * TaskUmbilicalProtocol which was originally
+ * a {@code WritableRPC} protocol.
+ * To marshall some of the complex MR datatypes, such as {@code Task},
+ * some objects are serialized to bytes
+ * and send in the RPC messages as binary objects, rather than fully
+ * defined in protobuf messages.
+ */
-class JvmContext implements Writable {
[email protected]
+package org.apache.hadoop.mapred.protocolPB;
- public static final Logger LOG =
- LoggerFactory.getLogger(JvmContext.class);
-
- JVMId jvmId;
- String pid;
-
- JvmContext() {
- jvmId = new JVMId();
- pid = "";
- }
-
- JvmContext(JVMId id, String pid) {
- jvmId = id;
- this.pid = pid;
- }
-
- public void readFields(DataInput in) throws IOException {
- jvmId.readFields(in);
- this.pid = Text.readString(in);
- }
-
- public void write(DataOutput out) throws IOException {
- jvmId.write(out);
- Text.writeString(out, pid);
- }
-}
+import org.apache.hadoop.classification.InterfaceAudience;
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/security/authorize/MRAMPolicyProvider.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/security/authorize/MRAMPolicyProvider.java
index 24d0c2f7f92..c4ee101cf3f 100644
---
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/security/authorize/MRAMPolicyProvider.java
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/security/authorize/MRAMPolicyProvider.java
@@ -19,7 +19,7 @@
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.mapred.TaskUmbilicalProtocol;
+import org.apache.hadoop.mapred.protocolPB.TaskUmbilicalProtocolPB;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.v2.api.MRClientProtocolPB;
import org.apache.hadoop.security.authorize.PolicyProvider;
@@ -36,7 +36,7 @@ public class MRAMPolicyProvider extends PolicyProvider {
new Service[] {
new Service(
MRJobConfig.MR_AM_SECURITY_SERVICE_AUTHORIZATION_TASK_UMBILICAL,
- TaskUmbilicalProtocol.class),
+ TaskUmbilicalProtocolPB.class),
new Service(
MRJobConfig.MR_AM_SECURITY_SERVICE_AUTHORIZATION_CLIENT,
MRClientProtocolPB.class)
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/proto/TaskUmbilicalProtocol.proto
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/proto/TaskUmbilicalProtocol.proto
new file mode 100644
index 00000000000..61e683469f7
--- /dev/null
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/proto/TaskUmbilicalProtocol.proto
@@ -0,0 +1,187 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * These .proto interfaces are private and subject to change.
+ * Please see https://issues.apache.org/jira/browse/MAPREDUCE-6706
+ */
+
+syntax = "proto2";
+option java_package = "org.apache.hadoop.mapred.proto";
+option java_outer_classname = "TaskUmbilicalProtocolProtos";
+option java_generic_services = true;
+option java_generate_equals_and_hash = true;
+package hadoop.mapreduce.app;
+
+/*
+ Writable-serialized types are carried as opaque bytes, for example
jvm_context and
+ jvm_task.
+ TaskStatus is marshalled as a byte array which also includes a boolean to
declare
+ what subclass of TaskStatus to instantiate.
+ All parameter bytes fields are optional to handle null values.
+ */
+
+// getTask
+message GetTaskRequestProto {
+ optional bytes jvm_context = 1;
+}
+message GetTaskResponseProto {
+ optional bytes jvm_task = 1;
+}
+
+// statusUpdate
+message StatusUpdateRequestProto {
+ optional bytes task_id = 1;
+ optional bytes task_status = 2;
+}
+message StatusUpdateResponseProto {
+ required bool task_found = 1;
+ required bool preemption = 2;
+}
+
+// reportDiagnosticInfo
+message ReportDiagnosticInfoRequestProto {
+ optional bytes task_id = 1;
+ required string trace = 2;
+}
+message ReportDiagnosticInfoResponseProto {
+}
+
+// reportNextRecordRange
+message ReportNextRecordRangeRequestProto {
+ optional bytes task_id = 1;
+ optional bytes range = 2;
+}
+message ReportNextRecordRangeResponseProto {
+}
+
+// done
+message DoneRequestProto {
+ optional bytes task_id = 1;
+}
+message DoneResponseProto {
+}
+
+// commitPending
+message CommitPendingRequestProto {
+ optional bytes task_id = 1;
+ optional bytes task_status = 2;
+}
+message CommitPendingResponseProto {
+}
+
+// canCommit
+message CanCommitRequestProto {
+ optional bytes task_id = 1;
+}
+message CanCommitResponseProto {
+ required bool can_commit = 1;
+}
+
+// shuffleError
+message ShuffleErrorRequestProto {
+ optional bytes task_id = 1;
+ required string message = 2;
+}
+message ShuffleErrorResponseProto {
+}
+
+// fsError
+message FsErrorRequestProto {
+ optional bytes task_id = 1;
+ required string message = 2;
+}
+message FsErrorResponseProto {
+}
+
+// fatalError
+message FatalErrorRequestProto {
+ optional bytes task_id = 1;
+ required string message = 2;
+ required bool fast_fail = 3;
+}
+message FatalErrorResponseProto {
+}
+
+// getMapCompletionEvents
+message GetMapCompletionEventsRequestProto {
+ optional bytes job_id = 1;
+ required int32 from_index = 2;
+ required int32 max_locs = 3;
+ optional bytes task_attempt_id = 4;
+}
+message GetMapCompletionEventsResponseProto {
+ optional bytes events_update = 1;
+}
+
+// preempted
+message PreemptedRequestProto {
+ optional bytes task_id = 1;
+ optional bytes task_status = 2;
+}
+message PreemptedResponseProto {
+}
+
+// getCheckpointID
+message GetCheckpointIDRequestProto {
+ optional bytes task_id = 1;
+}
+message GetCheckpointIDResponseProto {
+ optional bytes checkpoint_id = 1;
+}
+
+// setCheckpointID
+message SetCheckpointIDRequestProto {
+ optional bytes task_id = 1;
+ optional bytes checkpoint_id = 2;
+}
+message SetCheckpointIDResponseProto {
+}
+
+/* Task Umbilical Protocol. */
+
+service TaskUmbilicalProtocolService {
+ rpc getTask(GetTaskRequestProto)
+ returns (GetTaskResponseProto);
+ rpc statusUpdate(StatusUpdateRequestProto)
+ returns (StatusUpdateResponseProto);
+ rpc reportDiagnosticInfo(ReportDiagnosticInfoRequestProto)
+ returns (ReportDiagnosticInfoResponseProto);
+ rpc reportNextRecordRange(ReportNextRecordRangeRequestProto)
+ returns (ReportNextRecordRangeResponseProto);
+ rpc done(DoneRequestProto)
+ returns (DoneResponseProto);
+ rpc commitPending(CommitPendingRequestProto)
+ returns (CommitPendingResponseProto);
+ rpc canCommit(CanCommitRequestProto)
+ returns (CanCommitResponseProto);
+ rpc shuffleError(ShuffleErrorRequestProto)
+ returns (ShuffleErrorResponseProto);
+ rpc fsError(FsErrorRequestProto)
+ returns (FsErrorResponseProto);
+ rpc fatalError(FatalErrorRequestProto)
+ returns (FatalErrorResponseProto);
+ rpc getMapCompletionEvents(GetMapCompletionEventsRequestProto)
+ returns (GetMapCompletionEventsResponseProto);
+ rpc preempted(PreemptedRequestProto)
+ returns (PreemptedResponseProto);
+ rpc getCheckpointID(GetCheckpointIDRequestProto)
+ returns (GetCheckpointIDResponseProto);
+ rpc setCheckpointID(SetCheckpointIDRequestProto)
+ returns (SetCheckpointIDResponseProto);
+}
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JvmContext.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JvmContext.java
index 88eeefc8e76..4b7d379fc8c 100644
---
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JvmContext.java
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JvmContext.java
@@ -27,7 +27,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-class JvmContext implements Writable {
+public class JvmContext implements Writable {
public static final Logger LOG =
LoggerFactory.getLogger(JvmContext.class);
@@ -35,12 +35,12 @@ class JvmContext implements Writable {
JVMId jvmId;
String pid;
- JvmContext() {
+ public JvmContext() {
jvmId = new JVMId();
pid = "";
}
- JvmContext(JVMId id, String pid) {
+ public JvmContext(JVMId id, String pid) {
jvmId = id;
this.pid = pid;
}
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/SortedRanges.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/SortedRanges.java
index 9d4e7354f0d..071f33e9e1a 100644
---
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/SortedRanges.java
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/SortedRanges.java
@@ -35,7 +35,7 @@
* Provides the SkipRangeIterator, which skips the Ranges
* stored in this object.
*/
-class SortedRanges implements Writable{
+public class SortedRanges implements Writable{
private static final Logger LOG =
LoggerFactory.getLogger(SortedRanges.class);
@@ -221,7 +221,7 @@ public String toString() {
* A Range can be of 0 length also. The Range stores indices
* of type long.
*/
- static class Range implements Comparable<Range>, Writable{
+ public static class Range implements Comparable<Range>, Writable{
private long startIndex;
private long length;
@@ -233,7 +233,7 @@ static class Range implements Comparable<Range>, Writable{
this.length = length;
}
- Range() {
+ public Range() {
this(0,0);
}
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TaskStatus.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TaskStatus.java
index 7496d7c6c25..88d67d6ff6d 100644
---
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TaskStatus.java
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TaskStatus.java
@@ -519,8 +519,52 @@ static TaskStatus createTaskStatus(boolean isMap,
TaskAttemptID taskId,
taskTracker, phase, counters);
}
+ /**
+ * Enum identifying the concrete type of a {@link TaskStatus} for
serialization.
+ */
+ public enum TaskStatusKind {
+ MapTaskStatus,
+ ReduceTaskStatus
+ }
+
static TaskStatus createTaskStatus(boolean isMap) {
- return (isMap) ? new MapTaskStatus() : new ReduceTaskStatus();
+ return createTaskStatus(
+ isMap ? TaskStatusKind.MapTaskStatus :
TaskStatusKind.ReduceTaskStatus);
+ }
+
+ static TaskStatus createTaskStatus(TaskStatusKind kind) {
+ return kind == TaskStatusKind.MapTaskStatus
+ ? new MapTaskStatus() : new ReduceTaskStatus();
+ }
+
+ /**
+ * Write a TaskStatus to a DataOutput, prefixed by a {@link TaskStatusKind}
discriminator.
+ * Used by the Protobuf-based RPC layer to serialize TaskStatus values.
+ * @param out output stream
+ * @param status status to write (must not be null)
+ * @throws IOException on I/O error
+ */
+ public static void writeTaskStatusForPB(DataOutput out, TaskStatus status)
+ throws IOException {
+ TaskStatusKind kind = status.getIsMap()
+ ? TaskStatusKind.MapTaskStatus : TaskStatusKind.ReduceTaskStatus;
+ out.writeByte(kind.ordinal());
+ status.write(out);
+ }
+
+ /**
+ * Read a TaskStatus that was written with {@link #writeTaskStatusForPB}.
+ * @param in input stream
+ * @return the deserialized TaskStatus
+ * @throws IOException on I/O error
+ */
+ public static TaskStatus readTaskStatusFromPB(DataInput in)
+ throws IOException {
+ int ordinal = in.readByte() & 0xFF;
+ TaskStatusKind kind = TaskStatusKind.values()[ordinal];
+ TaskStatus status = createTaskStatus(kind);
+ status.readFields(in);
+ return status;
}
}
diff --git
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/security/TestUmbilicalProtocolWithJobToken.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/security/TestUmbilicalProtocolWithJobToken.java
index 56f95469b39..d8fdd174f95 100644
---
a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/security/TestUmbilicalProtocolWithJobToken.java
+++
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/security/TestUmbilicalProtocolWithJobToken.java
@@ -19,12 +19,11 @@
package org.apache.hadoop.mapreduce.security;
import static
org.apache.hadoop.fs.CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION;
-import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.doReturn;
+import java.io.Closeable;
import java.net.InetSocketAddress;
import java.security.PrivilegedExceptionAction;
@@ -33,10 +32,15 @@
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.ipc.Client;
+import org.apache.hadoop.ipc.ProtobufRpcEngine2;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.Server;
-import org.apache.hadoop.ipc.ProtocolSignature;
+import org.apache.hadoop.mapred.AMFeedback;
import org.apache.hadoop.mapred.TaskUmbilicalProtocol;
+import org.apache.hadoop.mapred.proto.TaskUmbilicalProtocolProtos;
+import org.apache.hadoop.mapred.protocolPB.TaskUmbilicalProtocolPB;
+import org.apache.hadoop.mapred.protocolPB.TaskUmbilicalProtocolPBClientImpl;
+import
org.apache.hadoop.mapred.protocolPB.TaskUmbilicalProtocolServerSideTranslatorPB;
import org.apache.hadoop.mapreduce.security.token.JobTokenIdentifier;
import org.apache.hadoop.mapreduce.security.token.JobTokenSecretManager;
import org.apache.hadoop.net.NetUtils;
@@ -51,10 +55,10 @@
import org.junit.jupiter.api.Test;
import static org.slf4j.LoggerFactory.getLogger;
-/** Unit tests for using Job Token over RPC.
- *
+/** Unit tests for using Job Token over RPC.
+ *
* System properties required:
- *
-Djava.security.krb5.conf=.../hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/target/test-classes/krb5.conf
+ *
-Djava.security.krb5.conf=.../hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/target/test-classes/krb5.conf
* -Djava.net.preferIPv4Stack=true
*/
public class TestUmbilicalProtocolWithJobToken {
@@ -80,16 +84,22 @@ public class TestUmbilicalProtocolWithJobToken {
@Test
public void testJobTokenRpc() throws Exception {
TaskUmbilicalProtocol mockTT = mock(TaskUmbilicalProtocol.class);
- doReturn(TaskUmbilicalProtocol.versionID)
- .when(mockTT).getProtocolVersion(anyString(), anyLong());
- doReturn(ProtocolSignature.getProtocolSignature(
- mockTT, TaskUmbilicalProtocol.class.getName(),
- TaskUmbilicalProtocol.versionID, 0))
- .when(mockTT).getProtocolSignature(anyString(), anyLong(), anyInt());
+ AMFeedback feedback = new AMFeedback();
+ feedback.setTaskFound(true);
+ doReturn(feedback).when(mockTT).statusUpdate(any(), any());
+
+ TaskUmbilicalProtocolServerSideTranslatorPB translator =
+ new TaskUmbilicalProtocolServerSideTranslatorPB(mockTT);
+ org.apache.hadoop.thirdparty.protobuf.BlockingService blockingService =
+ TaskUmbilicalProtocolProtos.TaskUmbilicalProtocolService
+ .newReflectiveBlockingService(translator);
JobTokenSecretManager sm = new JobTokenSecretManager();
+ RPC.setProtocolEngine(conf, TaskUmbilicalProtocolPB.class,
+ ProtobufRpcEngine2.class);
final Server server = new RPC.Builder(conf)
- .setProtocol(TaskUmbilicalProtocol.class).setInstance(mockTT)
+ .setProtocol(TaskUmbilicalProtocolPB.class)
+ .setInstance(blockingService)
.setBindAddress(ADDRESS).setPort(0).setNumHandlers(5).setVerbose(true)
.setSecretManager(sm).build();
@@ -109,14 +119,16 @@ public void testJobTokenRpc() throws Exception {
public Object run() throws Exception {
TaskUmbilicalProtocol proxy = null;
try {
- proxy = (TaskUmbilicalProtocol) RPC.getProxy(
- TaskUmbilicalProtocol.class, TaskUmbilicalProtocol.versionID,
+ TaskUmbilicalProtocolPB pbProxy = RPC.getProxy(
+ TaskUmbilicalProtocolPB.class,
+ RPC.getProtocolVersion(TaskUmbilicalProtocolPB.class),
addr, conf);
+ proxy = new TaskUmbilicalProtocolPBClientImpl(pbProxy);
proxy.statusUpdate(null, null);
} finally {
server.stop();
- if (proxy != null) {
- RPC.stopProxy(proxy);
+ if (proxy instanceof Closeable closeable) {
+ closeable.close();
}
}
return null;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
