From 28759a7f4c021dfa7b23d21597fa0ad229169862 Mon Sep 17 00:00:00 2001 From: Laszlo Bodor Date: Tue, 2 Jun 2026 11:51:55 +0200 Subject: [PATCH 1/2] TEZ-4723: TestMapProcessor is flaky --- .../processor/map/TestMapProcessor.java | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java index 556968de66..d87632f850 100644 --- a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java +++ b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java @@ -25,6 +25,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; @@ -34,6 +35,10 @@ import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.MapReduceBase; +import org.apache.hadoop.mapred.Mapper; +import org.apache.hadoop.mapred.OutputCollector; +import org.apache.hadoop.mapred.Reporter; import org.apache.hadoop.mapreduce.MRConfig; import org.apache.tez.common.MRFrameworkConfigs; import org.apache.tez.common.TezRuntimeFrameworkConfigs; @@ -197,6 +202,30 @@ public void testMapProcessor() throws Exception { reader.close(); } + /** + * A mapper that sleeps briefly every N records to ensure progress updates + * can be captured by the test's monitoring thread. + */ + public static class SlowMapper extends MapReduceBase + implements Mapper { + private static final int N = 100; + private final AtomicInteger counter = new AtomicInteger(0); + + @Override + public void map(LongWritable key, Text value, + OutputCollector output, Reporter reporter) + throws IOException { + output.collect(key, value); + if (counter.incrementAndGet() % N == 0) { + try { + Thread.sleep(1); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + } + @Test(timeout = 30000) public void testMapProcessorProgress() throws Exception { String dagName = "mrdag0"; @@ -204,6 +233,8 @@ public void testMapProcessorProgress() throws Exception { JobConf jobConf = new JobConf(defaultConf); setUpJobConf(jobConf); + jobConf.setMapperClass(SlowMapper.class); + MRHelpers.translateMRConfToTez(jobConf); jobConf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID, 0); From 2b9f69d2f6ba33cd54c2b20a07ec0eb6b6326830 Mon Sep 17 00:00:00 2001 From: Laszlo Bodor Date: Wed, 3 Jun 2026 13:28:17 +0200 Subject: [PATCH 2/2] LatchMapper --- .../processor/map/TestMapProcessor.java | 47 ++++++++++--------- 1 file changed, 24 insertions(+), 23 deletions(-) diff --git a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java index d87632f850..92cb1c36b5 100644 --- a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java +++ b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java @@ -22,10 +22,10 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.Collections; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; @@ -40,6 +40,7 @@ import org.apache.hadoop.mapred.OutputCollector; import org.apache.hadoop.mapred.Reporter; import org.apache.hadoop.mapreduce.MRConfig; +import org.apache.hadoop.test.GenericTestUtils; import org.apache.tez.common.MRFrameworkConfigs; import org.apache.tez.common.TezRuntimeFrameworkConfigs; import org.apache.tez.common.TezSharedExecutor; @@ -67,7 +68,6 @@ import org.apache.tez.runtime.library.output.OrderedPartitionedKVOutput; import org.junit.After; -import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.slf4j.Logger; @@ -81,7 +81,7 @@ public class TestMapProcessor { private static JobConf defaultConf = new JobConf(); private static FileSystem localFs = null; private static Path workDir = null; - static float progressUpdate = 0.0f; + private static CountDownLatch progressLatch; static { try { defaultConf.set("fs.defaultFS", "file:///"); @@ -203,22 +203,23 @@ public void testMapProcessor() throws Exception { } /** - * A mapper that sleeps briefly every N records to ensure progress updates - * can be captured by the test's monitoring thread. + * A mapper that waits on a latch after the first record, giving the + * ProgressHelper time to report intermediate progress. The latch is + * released by the test once progress has been captured. */ - public static class SlowMapper extends MapReduceBase + public static class LatchMapper extends MapReduceBase implements Mapper { - private static final int N = 100; - private final AtomicInteger counter = new AtomicInteger(0); + private boolean waited = false; @Override public void map(LongWritable key, Text value, OutputCollector output, Reporter reporter) throws IOException { output.collect(key, value); - if (counter.incrementAndGet() % N == 0) { + if (!waited) { + waited = true; try { - Thread.sleep(1); + progressLatch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } @@ -233,7 +234,8 @@ public void testMapProcessorProgress() throws Exception { JobConf jobConf = new JobConf(defaultConf); setUpJobConf(jobConf); - jobConf.setMapperClass(SlowMapper.class); + jobConf.setMapperClass(LatchMapper.class); + progressLatch = new CountDownLatch(1); MRHelpers.translateMRConfToTez(jobConf); jobConf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID, 0); @@ -246,7 +248,7 @@ public void testMapProcessorProgress() throws Exception { Path mapInput = new Path(workDir, "map0"); - MapUtils.generateInputSplit(localFs, workDir, jobConf, mapInput, 100000); + MapUtils.generateInputSplit(localFs, workDir, jobConf, mapInput, 100); InputSpec mapInputSpec = new InputSpec("NullSrcVertex", InputDescriptor.create(MRInputLegacy.class.getName()) @@ -268,21 +270,20 @@ public void testMapProcessorProgress() throws Exception { Collections.singletonList(mapOutputSpec), sharedExecutor); ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); - Thread monitorProgress = new Thread(new Runnable() { - @Override - public void run() { - float prog = task.getProgress(); - if(prog > 0.0f && prog < 1.0f) - progressUpdate = prog; + scheduler.scheduleAtFixedRate(() -> { + float prog = task.getProgress(); + if (prog > 0.0f && prog < 1.0f) { + progressLatch.countDown(); } - }); + }, 0, 1, TimeUnit.MILLISECONDS); task.initialize(); - scheduler.scheduleAtFixedRate(monitorProgress, 0, 1, - TimeUnit.MILLISECONDS); task.run(); - Assert.assertTrue("Progress Updates should be captured!", - progressUpdate > 0.0f && progressUpdate < 1.0f); + + GenericTestUtils.waitFor(() -> progressLatch.getCount() == 0, 10, 5000, + "Progress update between 0.0 and 1.0 was never captured"); + + scheduler.shutdownNow(); task.close(); sharedExecutor.shutdownNow(); }