Skip to content

MAPREDUCE-7369. Fixed MapReduce tasks timing out when spends more time on MultipleOutputs#close #4247

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
iwasakims merged 6 commits into from
Jun 20, 2022
Merged

MAPREDUCE-7369. Fixed MapReduce tasks timing out when spends more time on MultipleOutputs#close #4247

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
aa1e7fa
MAPREDUCE-7369. Fixed MapReduce tasks timing out when spends more tim…
Apr 29, 2022
c168a39
Removed redundant public static final
Apr 29, 2022
9c969c3
Updating DEFAULT_MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK to true
May 3, 2022
f242db0
Removing config for liveness check and making it default
May 6, 2022
39b2ace
Disabling the ping check feature by default and added a test case for…
Jun 19, 2022
592596b
update mapreduce.task.enable.ping-for-liveliness-check to mapreduce.t…
Jun 19, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 10 additions & 5 deletions ...-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/TaskAttemptListenerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,10 @@
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicReference;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hadoop.classification.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.ipc.ProtocolSignature;
Expand All @@ -50,8 +54,8 @@
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEventType;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptFailEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptStatusUpdateEvent.TaskAttemptStatus;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptStatusUpdateEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptStatusUpdateEvent.TaskAttemptStatus;
import org.apache.hadoop.mapreduce.v2.app.rm.RMHeartbeatHandler;
import org.apache.hadoop.mapreduce.v2.app.rm.preemption.AMPreemptionPolicy;
import org.apache.hadoop.mapreduce.v2.app.security.authorize.MRAMPolicyProvider;
Expand All @@ -61,10 +65,6 @@
import org.apache.hadoop.util.StringInterner;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hadoop.classification.VisibleForTesting;

/**
* This class is responsible for talking to the task umblical.
Expand Down Expand Up @@ -409,6 +409,11 @@ public AMFeedback statusUpdate(TaskAttemptID taskAttemptID,
if (LOG.isDebugEnabled()) {
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,
MRJobConfig.DEFAULT_MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK)) {
taskHeartbeatHandler.progressing(yarnAttemptID);
}
return feedback;
}

Expand Down
46 changes: 29 additions & 17 deletions ...reduce-client-app/src/test/java/org/apache/hadoop/mapred/TestTaskAttemptListenerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,23 +17,30 @@
*/
package org.apache.hadoop.mapred;

import java.util.function.Supplier;
import org.apache.hadoop.mapred.Counters.Counter;
import org.apache.hadoop.mapreduce.checkpoint.EnumCounter;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;

import org.junit.After;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.Counters.Counter;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.TaskType;
import org.apache.hadoop.mapreduce.TypeConverter;
import org.apache.hadoop.mapreduce.checkpoint.CheckpointID;
import org.apache.hadoop.mapreduce.checkpoint.EnumCounter;
import org.apache.hadoop.mapreduce.checkpoint.FSCheckpointID;
import org.apache.hadoop.mapreduce.checkpoint.TaskCheckpointID;
import org.apache.hadoop.mapreduce.security.token.JobTokenSecretManager;
Expand All @@ -48,9 +55,9 @@
import org.apache.hadoop.mapreduce.v2.app.job.Job;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptStatusUpdateEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptStatusUpdateEvent.TaskAttemptStatus;
import org.apache.hadoop.mapreduce.v2.app.rm.RMHeartbeatHandler;
import org.apache.hadoop.mapreduce.v2.app.rm.preemption.AMPreemptionPolicy;
import org.apache.hadoop.mapreduce.v2.app.rm.preemption.CheckpointAMPreemptionPolicy;
import org.apache.hadoop.mapreduce.v2.app.rm.RMHeartbeatHandler;
import org.apache.hadoop.mapreduce.v2.util.MRBuilderUtils;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.yarn.event.Dispatcher;
Expand All @@ -60,17 +67,22 @@
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.util.ControlledClock;
import org.apache.hadoop.yarn.util.SystemClock;
import org.junit.After;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

/**
* Tests the behavior of TaskAttemptListenerImpl.
Expand Down Expand Up @@ -403,18 +415,18 @@ public void testStatusUpdateProgress()
startListener(true);
verify(hbHandler).register(attemptId);

// make sure a ping doesn't report progress
// make sure a ping does report progress
AMFeedback feedback = listener.statusUpdate(attemptID, null);
assertTrue(feedback.getTaskFound());
verify(hbHandler, never()).progressing(eq(attemptId));
verify(hbHandler, times(1)).progressing(eq(attemptId));

// make sure a status update does report progress
MapTaskStatus mockStatus = new MapTaskStatus(attemptID, 0.0f, 1,
TaskStatus.State.RUNNING, "", "RUNNING", "", TaskStatus.Phase.MAP,
new Counters());
feedback = listener.statusUpdate(attemptID, mockStatus);
assertTrue(feedback.getTaskFound());
verify(hbHandler).progressing(eq(attemptId));
verify(hbHandler, times(2)).progressing(eq(attemptId));
}

@Test
Expand Down
6 changes: 6 additions & 0 deletions ...t/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -919,6 +919,12 @@ public interface MRJobConfig {
MR_AM_PREFIX + "scheduler.heartbeat.interval-ms";
public static final int DEFAULT_MR_AM_TO_RM_HEARTBEAT_INTERVAL_MS = 1000;

/** Whether to consider ping from tasks in liveliness check. */
String MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK =
"mapreduce.task.enable.ping-for-liveliness-check";
boolean DEFAULT_MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK
= false;

/**
* If contact with RM is lost, the AM will wait MR_AM_TO_RM_WAIT_INTERVAL_MS
* milliseconds before aborting. During this interval, AM will still try
Expand Down
7 changes: 7 additions & 0 deletions ...adoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -286,6 +286,13 @@
</description>
</property>

<property>
<name>mapreduce.task.enable.ping-for-liveliness-check</name>
<value>true</value>

@iwasakims iwasakims May 3, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The value must be same with the DEFAULT_MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK.

@hotcodemacha hotcodemacha May 3, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @iwasakims for review. I have updated the DEFAULT_MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK to true.

@iwasakims iwasakims May 5, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When should we turn this off? If it should be always true, we don't need to make it configurable. We usually needs new configuration knob to add new feature disabled by default for compatibility and safety.

@hotcodemacha hotcodemacha May 5, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah I think this makes sense, I will remove configurable part.

@hotcodemacha hotcodemacha May 14, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @iwasakims. I saw the comment from @cnauroth on JIRA about his suggestion to put it behind the configuration and I think we can keep it configurable and let the end user decide on it. By default, lets keep it

mapreduce.task.enable.ping-for-liveliness-check : false which will keep the current behaviour intact

and for use cases that we saw in JIRA, user can set it true when required

Thoughts?

@iwasakims iwasakims Jun 1, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ashutoshcipher Making this feature disabled by default sounds right direction as @cnauroth mentioned. While no unexpected side effect was observed in the test results of current patch, there should be additional test case covering the code path of new feature enabled.

@hotcodemacha hotcodemacha Jun 2, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @iwasakims , let me make the required changes.

@hotcodemacha hotcodemacha Jun 19, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@iwasakims @cnauroth - I have made the required changes, Can you please review the PR? Thanks

<description>Whether to consider ping from tasks in liveliness check.
</description>
</property>

<property>
<name>mapreduce.map.memory.mb</name>
<value>-1</value>
Expand Down