From 0a695ba4c1ec0da0701e32e5c3c6b5a33fda4ba1 Mon Sep 17 00:00:00 2001 From: Denys Kuzmenko Date: Wed, 15 Jul 2026 16:40:39 +0300 Subject: [PATCH 1/2] TEZ-4737: Session DAG submission fails with FileAlreadyExistsException when the session outlives the TezClient --- .../java/org/apache/tez/client/TezClient.java | 4 +- .../org/apache/tez/client/TestTezClient.java | 61 ++++++++++++++++--- ...GClientAMProtocolBlockingPBServerImpl.java | 11 ++++ ...GClientAMProtocolBlockingPBServerImpl.java | 4 ++ 4 files changed, 69 insertions(+), 11 deletions(-) diff --git a/tez-api/src/main/java/org/apache/tez/client/TezClient.java b/tez-api/src/main/java/org/apache/tez/client/TezClient.java index 74942ac06d..2fd30aeff2 100644 --- a/tez-api/src/main/java/org/apache/tez/client/TezClient.java +++ b/tez-api/src/main/java/org/apache/tez/client/TezClient.java @@ -692,7 +692,9 @@ private DAGClient submitDAGSession(DAG dag) throws TezException, IOException { serializedSubmitDAGPlanRequestCounter.incrementAndGet()); FileSystem fs = dagPlanPath.getFileSystem(stagingFs.getConf()); - try (FSDataOutputStream fsDataOutputStream = fs.create(dagPlanPath, false)) { + // Overwrite a possible leftover plan file: when the session outlives the TezClient instances, + // a file with the same name may have been left behind by an already consumed submission. + try (FSDataOutputStream fsDataOutputStream = fs.create(dagPlanPath, true)) { LOG.info("Send dag plan using YARN local resources since it's too large" + ", dag plan size=" + request.getSerializedSize() + ", max dag plan size through IPC=" + maxSubmitDAGRequestSizeThroughIPC diff --git a/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java b/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java index 19d0668175..6a2912ab5a 100644 --- a/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java +++ b/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java @@ -70,6 +70,7 @@ import org.apache.hadoop.yarn.client.api.YarnClient; import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException; import org.apache.hadoop.yarn.exceptions.YarnException; +import org.apache.tez.common.TezCommonUtils; import org.apache.tez.common.counters.LimitExceededException; import org.apache.tez.common.counters.Limits; import org.apache.tez.common.counters.TezCounters; @@ -269,20 +270,12 @@ private void _testTezClientSessionLargeDAGPlan(int maxIPCMsgSize, int payloadSiz localResourceMap.put(lrName, LocalResource.newInstance(URL.newInstance("file", "localhost", 0, "/test"), LocalResourceType.FILE, LocalResourceVisibility.PUBLIC, 1, 1)); - ProcessorDescriptor processorDescriptor = ProcessorDescriptor.create("P"); - processorDescriptor.setUserPayload(UserPayload.create(ByteBuffer.allocate(payloadSize))); - Vertex vertex = Vertex.create("Vertex", processorDescriptor, 1, Resource.newInstance(1, 1)); - DAG dag = DAG.create("DAG-testTezClientSessionLargeDAGPlan").addVertex(vertex); - client.start(); client.addAppMasterLocalFiles(localResourceMap); - client.submitDAG(dag); + SubmitDAGRequestProto request = + submitDAGAndCaptureRequest(client, largeDAG("DAG-testTezClientSessionLargeDAGPlan", payloadSize)); client.stop(); - ArgumentCaptor captor = ArgumentCaptor.forClass(SubmitDAGRequestProto.class); - verify(client.sessionAmProxy).submitDAG(any(), captor.capture()); - SubmitDAGRequestProto request = captor.getValue(); - if (shouldSerialize) { /* we need manually delete the serialized dagplan since staging path here won't be destroyed */ Path dagPlanPath = new Path(request.getSerializedRequestPath()); @@ -300,6 +293,54 @@ private void _testTezClientSessionLargeDAGPlan(int maxIPCMsgSize, int payloadSiz } } + @Test + @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS) + public void testSessionLargeDAGPlanWithLeftoverPlanFile() throws Exception { + // Simulates long-running/external sessions (e.g. Hive with external session pools): the + // application and its staging dir outlive the TezClient instances, which are recreated per + // query with their serialized-plan-file counter restarting from 0. A plan file left behind + // by a previous client generation must not fail subsequent submissions with + // FileAlreadyExistsException. + TezConfiguration conf = new TezConfiguration(); + conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, 1024 * 1024); + conf.set(TezConfiguration.TEZ_AM_STAGING_DIR, STAGING_DIR.getAbsolutePath()); + + Path baseStagingPath = TezCommonUtils.getTezBaseStagingPath(conf); + FileSystem localFs = FileSystem.getLocal(conf); + localFs.delete(baseStagingPath, true); + + // first client generation writes tez-dag.pb1 into the session's staging dir and is then + // abandoned without stop() - the session and its staging dir keep running + TezClientForTest client1 = configureAndCreateTezClient(null, true, conf); + client1.start(); + submitDAGAndCaptureRequest(client1, largeDAG("DAG-gen1", 2 * 1024 * 1024)); + + // second client generation reconnects to the same session: same appId, same staging dir, + // restarted plan-file counter - it computes the same tez-dag.pb1 path + TezClientForTest client2 = configureAndCreateTezClient(null, true, conf); + client2.start(); + SubmitDAGRequestProto request = submitDAGAndCaptureRequest(client2, largeDAG("DAG-gen2", 2 * 1024 * 1024)); + client2.stop(); + + assertTrue(request.hasSerializedRequestPath()); + + localFs.delete(baseStagingPath, true); + } + + private DAG largeDAG(String name, int payloadSize) { + ProcessorDescriptor processorDescriptor = ProcessorDescriptor.create("P"); + processorDescriptor.setUserPayload(UserPayload.create(ByteBuffer.allocate(payloadSize))); + Vertex vertex = Vertex.create("Vertex", processorDescriptor, 1, Resource.newInstance(1, 1)); + return DAG.create(name).addVertex(vertex); + } + + private SubmitDAGRequestProto submitDAGAndCaptureRequest(TezClientForTest client, DAG dag) throws Exception { + client.submitDAG(dag); + ArgumentCaptor captor = ArgumentCaptor.forClass(SubmitDAGRequestProto.class); + verify(client.sessionAmProxy).submitDAG(any(), captor.capture()); + return captor.getValue(); + } + @Test @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS) public void testGetClient() throws Exception { diff --git a/tez-dag/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientAMProtocolBlockingPBServerImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientAMProtocolBlockingPBServerImpl.java index 1b7601cb6b..c5da19477a 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientAMProtocolBlockingPBServerImpl.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientAMProtocolBlockingPBServerImpl.java @@ -58,8 +58,13 @@ import com.google.protobuf.RpcController; import com.google.protobuf.ServiceException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + public class DAGClientAMProtocolBlockingPBServerImpl implements DAGClientAMProtocolBlockingPB { + private static final Logger LOG = LoggerFactory.getLogger(DAGClientAMProtocolBlockingPBServerImpl.class); + DAGClientHandler real; final FileSystem stagingFs; @@ -176,6 +181,12 @@ public SubmitDAGResponseProto submitDAG(RpcController controller, request = SubmitDAGRequestProto.parseFrom(in); } catch (IOException e) { throw wrapException(e); + } finally { + try { + fs.delete(requestPath, false); + } catch (IOException e) { + LOG.warn("Failed to delete the serialized DAG plan file {}", requestPath, e); + } } } DAGPlan dagPlan = request.getDAGPlan(); diff --git a/tez-dag/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClientAMProtocolBlockingPBServerImpl.java b/tez-dag/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClientAMProtocolBlockingPBServerImpl.java index 360d5ac138..9878d321b3 100644 --- a/tez-dag/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClientAMProtocolBlockingPBServerImpl.java +++ b/tez-dag/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClientAMProtocolBlockingPBServerImpl.java @@ -20,6 +20,7 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.any; import static org.mockito.Mockito.mock; @@ -118,6 +119,9 @@ public void testSubmitDagInSessionWithLargeDagPlan() throws Exception { requestBuilder.clear().setSerializedRequestPath(requestFile.getAbsolutePath()); serverImpl.submitDAG(null, requestBuilder.build()); + assertFalse(requestFile.exists(), + "The serialized DAG plan file should be deleted once the AM has consumed it"); + ArgumentCaptor dagPlanCaptor = ArgumentCaptor.forClass(DAGPlan.class); verify(dagClientHandler).submitDAG(dagPlanCaptor.capture(), localResourcesCaptor.capture()); dagPlan = dagPlanCaptor.getValue(); From 3ee5c295056e2ff444feb8ffc526b6f09e54f570 Mon Sep 17 00:00:00 2001 From: Denys Kuzmenko Date: Thu, 16 Jul 2026 19:14:17 +0300 Subject: [PATCH 2/2] review comments #1 --- .../java/org/apache/tez/client/TezClient.java | 6 ++--- .../org/apache/tez/dag/api/TezConstants.java | 1 + .../org/apache/tez/client/TestTezClient.java | 27 ++++++++++--------- 3 files changed, 17 insertions(+), 17 deletions(-) diff --git a/tez-api/src/main/java/org/apache/tez/client/TezClient.java b/tez-api/src/main/java/org/apache/tez/client/TezClient.java index 2fd30aeff2..474c181a33 100644 --- a/tez-api/src/main/java/org/apache/tez/client/TezClient.java +++ b/tez-api/src/main/java/org/apache/tez/client/TezClient.java @@ -688,12 +688,10 @@ private DAGClient submitDAGSession(DAG dag) throws TezException, IOException { SubmitDAGRequestProto request = requestBuilder.build(); if (request.getSerializedSize() > maxSubmitDAGRequestSizeThroughIPC) { Path dagPlanPath = new Path(TezCommonUtils.getTezSystemStagingPath(amConfig.getTezConfiguration(), - sessionAppId.toString()), TezConstants.TEZ_PB_PLAN_BINARY_NAME + - serializedSubmitDAGPlanRequestCounter.incrementAndGet()); + sessionAppId.toString()), TezConstants.TEZ_PB_PLAN_BINARY_NAME_FORMAT.formatted( + serializedSubmitDAGPlanRequestCounter.incrementAndGet())); FileSystem fs = dagPlanPath.getFileSystem(stagingFs.getConf()); - // Overwrite a possible leftover plan file: when the session outlives the TezClient instances, - // a file with the same name may have been left behind by an already consumed submission. try (FSDataOutputStream fsDataOutputStream = fs.create(dagPlanPath, true)) { LOG.info("Send dag plan using YARN local resources since it's too large" + ", dag plan size=" + request.getSerializedSize() diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java b/tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java index def923c9b3..2031a20003 100644 --- a/tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java +++ b/tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java @@ -48,6 +48,7 @@ public final class TezConstants { public static final String SERVICE_PLUGINS_DESCRIPTOR_JSON = "service_plugins_descriptor.json"; public static final String TEZ_PB_BINARY_CONF_NAME = "tez-conf.pb"; public static final String TEZ_PB_PLAN_BINARY_NAME = "tez-dag.pb"; + public static final String TEZ_PB_PLAN_BINARY_NAME_FORMAT = "tez-dag-%d.pb"; public static final String TEZ_PB_PLAN_TEXT_NAME = "tez-dag.pb.txt"; /* diff --git a/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java b/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java index 6a2912ab5a..402c7a545e 100644 --- a/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java +++ b/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java @@ -296,11 +296,8 @@ private void _testTezClientSessionLargeDAGPlan(int maxIPCMsgSize, int payloadSiz @Test @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS) public void testSessionLargeDAGPlanWithLeftoverPlanFile() throws Exception { - // Simulates long-running/external sessions (e.g. Hive with external session pools): the - // application and its staging dir outlive the TezClient instances, which are recreated per - // query with their serialized-plan-file counter restarting from 0. A plan file left behind - // by a previous client generation must not fail subsequent submissions with - // FileAlreadyExistsException. + // The session outlives its TezClient instances: a plan file left behind by a previous + // client must not fail subsequent submissions. TezConfiguration conf = new TezConfiguration(); conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, 1024 * 1024); conf.set(TezConfiguration.TEZ_AM_STAGING_DIR, STAGING_DIR.getAbsolutePath()); @@ -309,21 +306,25 @@ public void testSessionLargeDAGPlanWithLeftoverPlanFile() throws Exception { FileSystem localFs = FileSystem.getLocal(conf); localFs.delete(baseStagingPath, true); - // first client generation writes tez-dag.pb1 into the session's staging dir and is then - // abandoned without stop() - the session and its staging dir keep running + // client1 writes tez-dag-1.pb and is abandoned without stop() TezClientForTest client1 = configureAndCreateTezClient(null, true, conf); client1.start(); - submitDAGAndCaptureRequest(client1, largeDAG("DAG-gen1", 2 * 1024 * 1024)); + SubmitDAGRequestProto request1 = submitDAGAndCaptureRequest(client1, largeDAG("DAG-client1", 2 * 1024 * 1024)); + assertTrue(request1.hasSerializedRequestPath()); + // the mocked AM never consumes the plan file, leaving it behind in the staging dir + Path leftoverPlanPath = new Path(request1.getSerializedRequestPath()); + assertTrue(localFs.exists(leftoverPlanPath)); - // second client generation reconnects to the same session: same appId, same staging dir, - // restarted plan-file counter - it computes the same tez-dag.pb1 path + // client2 reuses the same session and computes the same tez-dag-1.pb path TezClientForTest client2 = configureAndCreateTezClient(null, true, conf); client2.start(); - SubmitDAGRequestProto request = submitDAGAndCaptureRequest(client2, largeDAG("DAG-gen2", 2 * 1024 * 1024)); + SubmitDAGRequestProto request2 = submitDAGAndCaptureRequest(client2, largeDAG("DAG-client2", 2 * 1024 * 1024)); + assertTrue(request2.hasSerializedRequestPath()); + // path collision: client2 overwrote client1's leftover instead of failing + assertEquals(leftoverPlanPath, new Path(request2.getSerializedRequestPath())); + assertTrue(localFs.exists(leftoverPlanPath)); client2.stop(); - assertTrue(request.hasSerializedRequestPath()); - localFs.delete(baseStagingPath, true); }