abstractdog commented on code in PR #519:
URL: https://github.com/apache/tez/pull/519#discussion_r3673389043
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestHistoryParser.java:
##########
@@ -209,22 +209,22 @@ public void testParserWithSuccessfulJob() throws
Exception {
String dagId = runWordCount(WordCount.TokenProcessor.class.getName(),
WordCount.SumProcessor.class.getName(), "WordCount", true);
- //Export the data from ATS
- String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR,
"--yarnTimelineAddress=" + yarnTimelineAddress };
+ //Retry the ATS export+parse pipeline until the resulting DagInfo actually
contains
+ //the expected DAG (two vertices, non-empty vertices/tasks). Under load
the AM's async
+ //flush and the timeline server's write path can race the export, leaving
empty/partial
+ //entities in the zip. Before TEZ-4733 that produced a misleading
+ //"A JSONObject text must begin with '{'" JSONException at parse time.
Review Comment:
this is not needed: "Before TEZ-4733 that produced a misleading
//"A JSONObject text must begin with '{'" JSONException at parse time."
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestHistoryParser.java:
##########
@@ -234,6 +234,77 @@ public void testParserWithSuccessfulJob() throws Exception
{
isDAGEqual(dagInfoFromATS, shDagInfo);
}
+ /**
+ * the ATS write path is async (AM event queue → timeline client →
+ * timeline server). Even after the DAG client reports the job complete,
timeline entities
+ * may still be in transit. Downloading too early produced empty zip entries
and a
+ * misleading JSONException: A JSONObject text must begin with '{'} at parse
time.
+ */
+ private DagInfo fetchDagInfoFromAtsWithRetry(String dagId, int maxAttempts,
+ long delayMs) throws Exception {
+ String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR,
+ "--yarnTimelineAddress=" + yarnTimelineAddress };
+ Exception lastError = null;
+ DagInfo lastPartial = null;
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+ try {
+ // Fresh download every attempt — ATSImportTool overwrites the zip.
+ int result = ATSImportTool.process(args);
+ assertEquals(0, result);
+ DagInfo info = getDagInfo(dagId);
+ if (isDagInfoComplete(info)) {
+ return info;
+ }
+ lastPartial = info;
+ } catch (Exception e) {
+ lastError = e;
+ }
+ if (attempt < maxAttempts) {
+ Thread.sleep(delayMs);
Review Comment:
I think it's overkill to provide `delayMs` and `maxAttempts` at the same time
use a reasonable `delayMs`, and let the user of this method define
`maxAttempts` accordingly
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestHistoryParser.java:
##########
@@ -234,6 +234,77 @@ public void testParserWithSuccessfulJob() throws Exception
{
isDAGEqual(dagInfoFromATS, shDagInfo);
}
+ /**
+ * the ATS write path is async (AM event queue → timeline client →
+ * timeline server). Even after the DAG client reports the job complete,
timeline entities
+ * may still be in transit. Downloading too early produced empty zip entries
and a
+ * misleading JSONException: A JSONObject text must begin with '{'} at parse
time.
+ */
+ private DagInfo fetchDagInfoFromAtsWithRetry(String dagId, int maxAttempts,
+ long delayMs) throws Exception {
+ String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR,
+ "--yarnTimelineAddress=" + yarnTimelineAddress };
+ Exception lastError = null;
+ DagInfo lastPartial = null;
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+ try {
+ // Fresh download every attempt — ATSImportTool overwrites the zip.
+ int result = ATSImportTool.process(args);
+ assertEquals(0, result);
+ DagInfo info = getDagInfo(dagId);
+ if (isDagInfoComplete(info)) {
+ return info;
+ }
+ lastPartial = info;
+ } catch (Exception e) {
+ lastError = e;
+ }
+ if (attempt < maxAttempts) {
+ Thread.sleep(delayMs);
+ }
+ }
+ fail("Could not fetch a complete DagInfo for " + dagId + " after " +
maxAttempts
+ + " attempts (lastError=" + lastError + ", lastPartial="
+ + (lastPartial == null ? "null"
+ : "vertices=" + lastPartial.getVertices().size()
+ + ", tasks=" + (lastPartial.getVertices().isEmpty() ? 0
+ :
lastPartial.getVertices().iterator().next().getTasks().size()))
+ + ")");
+ return null;
+ }
+
+ private static boolean isDagInfoComplete(DagInfo info) {
+ return info != null
+ && info.getVertices().size() >= 2
+ && info.getVertices().stream().allMatch(v ->
+ !v.getTasks().isEmpty()
+ && v.getTasks().stream().allMatch(t ->
!t.getTaskAttempts().isEmpty()));
+ }
+
+ private void waitForHistoryFileReady(String dagId, long timeoutMs) throws
Exception {
+ TezDAGID tezDAGID = TezDAGID.fromString(dagId);
+ ApplicationAttemptId applicationAttemptId =
ApplicationAttemptId.newInstance(tezDAGID
+ .getApplicationId(), 1);
+ Path historyPath = new Path(conf.get("fs.defaultFS")
+ + SIMPLE_HISTORY_DIR + HISTORY_TXT + "."
+ + applicationAttemptId);
Review Comment:
too many line breaks, it can be less I think
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestHistoryParser.java:
##########
@@ -234,6 +234,77 @@ public void testParserWithSuccessfulJob() throws Exception
{
isDAGEqual(dagInfoFromATS, shDagInfo);
}
+ /**
+ * the ATS write path is async (AM event queue → timeline client →
+ * timeline server). Even after the DAG client reports the job complete,
timeline entities
+ * may still be in transit. Downloading too early produced empty zip entries
and a
+ * misleading JSONException: A JSONObject text must begin with '{'} at parse
time.
+ */
+ private DagInfo fetchDagInfoFromAtsWithRetry(String dagId, int maxAttempts,
+ long delayMs) throws Exception {
+ String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR,
+ "--yarnTimelineAddress=" + yarnTimelineAddress };
+ Exception lastError = null;
+ DagInfo lastPartial = null;
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+ try {
+ // Fresh download every attempt — ATSImportTool overwrites the zip.
+ int result = ATSImportTool.process(args);
+ assertEquals(0, result);
+ DagInfo info = getDagInfo(dagId);
+ if (isDagInfoComplete(info)) {
+ return info;
+ }
+ lastPartial = info;
+ } catch (Exception e) {
+ lastError = e;
+ }
+ if (attempt < maxAttempts) {
+ Thread.sleep(delayMs);
+ }
+ }
+ fail("Could not fetch a complete DagInfo for " + dagId + " after " +
maxAttempts
+ + " attempts (lastError=" + lastError + ", lastPartial="
+ + (lastPartial == null ? "null"
+ : "vertices=" + lastPartial.getVertices().size()
+ + ", tasks=" + (lastPartial.getVertices().isEmpty() ? 0
+ :
lastPartial.getVertices().iterator().next().getTasks().size()))
+ + ")");
+ return null;
+ }
+
+ private static boolean isDagInfoComplete(DagInfo info) {
+ return info != null
+ && info.getVertices().size() >= 2
+ && info.getVertices().stream().allMatch(v ->
+ !v.getTasks().isEmpty()
+ && v.getTasks().stream().allMatch(t ->
!t.getTaskAttempts().isEmpty()));
+ }
+
+ private void waitForHistoryFileReady(String dagId, long timeoutMs) throws
Exception {
+ TezDAGID tezDAGID = TezDAGID.fromString(dagId);
+ ApplicationAttemptId applicationAttemptId =
ApplicationAttemptId.newInstance(tezDAGID
+ .getApplicationId(), 1);
+ Path historyPath = new Path(conf.get("fs.defaultFS")
+ + SIMPLE_HISTORY_DIR + HISTORY_TXT + "."
+ + applicationAttemptId);
+ FileSystem hfs = historyPath.getFileSystem(conf);
+ long deadline = System.currentTimeMillis() + timeoutMs;
+ long lastLen = -1L;
+ while (System.currentTimeMillis() < deadline) {
Review Comment:
use monothonic time
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestHistoryParser.java:
##########
@@ -234,6 +234,77 @@ public void testParserWithSuccessfulJob() throws Exception
{
isDAGEqual(dagInfoFromATS, shDagInfo);
}
+ /**
+ * the ATS write path is async (AM event queue → timeline client →
+ * timeline server). Even after the DAG client reports the job complete,
timeline entities
+ * may still be in transit. Downloading too early produced empty zip entries
and a
+ * misleading JSONException: A JSONObject text must begin with '{'} at parse
time.
+ */
+ private DagInfo fetchDagInfoFromAtsWithRetry(String dagId, int maxAttempts,
+ long delayMs) throws Exception {
+ String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR,
+ "--yarnTimelineAddress=" + yarnTimelineAddress };
+ Exception lastError = null;
+ DagInfo lastPartial = null;
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+ try {
+ // Fresh download every attempt — ATSImportTool overwrites the zip.
+ int result = ATSImportTool.process(args);
+ assertEquals(0, result);
+ DagInfo info = getDagInfo(dagId);
+ if (isDagInfoComplete(info)) {
+ return info;
+ }
+ lastPartial = info;
+ } catch (Exception e) {
+ lastError = e;
+ }
+ if (attempt < maxAttempts) {
+ Thread.sleep(delayMs);
+ }
+ }
+ fail("Could not fetch a complete DagInfo for " + dagId + " after " +
maxAttempts
+ + " attempts (lastError=" + lastError + ", lastPartial="
+ + (lastPartial == null ? "null"
+ : "vertices=" + lastPartial.getVertices().size()
+ + ", tasks=" + (lastPartial.getVertices().isEmpty() ? 0
+ :
lastPartial.getVertices().iterator().next().getTasks().size()))
+ + ")");
+ return null;
+ }
+
+ private static boolean isDagInfoComplete(DagInfo info) {
+ return info != null
+ && info.getVertices().size() >= 2
+ && info.getVertices().stream().allMatch(v ->
+ !v.getTasks().isEmpty()
+ && v.getTasks().stream().allMatch(t ->
!t.getTaskAttempts().isEmpty()));
Review Comment:
why is this check needed? is there a chance that the `DAGInfo` is
successfully parsed but it's not complete? Like: it contains fewer vertices
than expected? if so, `isDagInfoComplete` has to receive an
`expectedNumOfVertices` as a parameter for clarity's sake
##########
tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/ATSFileParser.java:
##########
@@ -174,11 +175,23 @@ private void processApplication(JSONObject
tezApplicationJson) throws JSONExcept
}
}
- private JSONObject readJson(InputStream in) throws IOException,
JSONException {
- //Read entire content to memory
- final NonSyncByteArrayOutputStream bout = new
NonSyncByteArrayOutputStream();
- IOUtils.copy(in, bout);
- return new JSONObject(new String(bout.toByteArray(), "UTF-8"));
+ /**
+ * Parse the raw payload of a single zip entry as JSON.
+ * Returns null if the payload is empty or blank — callers should skip such
entries.
+ */
+ private JSONObject readJson(byte[] payload, String entryName) throws
JSONException {
+ String text = new String(payload, StandardCharsets.UTF_8);
+ if (text.trim().isEmpty()) {
+ LOG.warn("Skipping zip entry '{}' - payload is empty or whitespace
only", entryName);
+ return null;
+ }
+ try {
+ return new JSONObject(text);
+ } catch (JSONException e) {
+ String snippet = text.length() > 200 ? text.substring(0, 200) + "..." :
text;
+ throw new JSONException("Failed to parse JSON from zip entry '" +
entryName
+ + "' (length=" + text.length() + ", snippet=" + snippet + "): " +
e.getMessage());
Review Comment:
this is for " enrich JSON parse errors with the offending entry name +
payload snippet", which makes sense to me
for clarity's sake what a JSONException was like before? was it really only
"A JSONObject text must begin with '{' at character 0 of "
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestHistoryParser.java:
##########
@@ -234,6 +234,77 @@ public void testParserWithSuccessfulJob() throws Exception
{
isDAGEqual(dagInfoFromATS, shDagInfo);
}
+ /**
+ * the ATS write path is async (AM event queue → timeline client →
+ * timeline server). Even after the DAG client reports the job complete,
timeline entities
+ * may still be in transit. Downloading too early produced empty zip entries
and a
+ * misleading JSONException: A JSONObject text must begin with '{'} at parse
time.
+ */
+ private DagInfo fetchDagInfoFromAtsWithRetry(String dagId, int maxAttempts,
+ long delayMs) throws Exception {
+ String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR,
+ "--yarnTimelineAddress=" + yarnTimelineAddress };
+ Exception lastError = null;
+ DagInfo lastPartial = null;
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
Review Comment:
it's usually rather `int attempt = 0; attempt < maxAttempts;`, isn't it?
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/parser/TestATSFileParser.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.tez.history.parser;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import org.apache.tez.dag.api.TezException;
+import org.apache.tez.history.parser.datamodel.DagInfo;
+
+import org.codehaus.jettison.json.JSONArray;
+import org.codehaus.jettison.json.JSONException;
+import org.codehaus.jettison.json.JSONObject;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class TestATSFileParser {
+
+ private static final String DAG_ID = "dag_1234567890_0001_1";
+
+ private static JSONObject minimalDagJson() throws JSONException {
+ JSONObject dag = new JSONObject();
+ dag.put("entityId", DAG_ID);
+ dag.put("entityType", "TEZ_DAG_ID");
+ JSONObject otherInfo = new JSONObject();
+ otherInfo.put("startTime", 1L);
+ otherInfo.put("endTime", 2L);
+ otherInfo.put("status", "SUCCEEDED");
+ otherInfo.put("counters", new JSONObject().put("counterGroups", new
JSONArray()));
+ dag.put("otherInfo", otherInfo);
+ return dag;
+ }
+
+ private static File writeZip(Path dir, String name, ZipContent... entries)
throws Exception {
+ File zip = dir.resolve(name).toFile();
+ try (FileOutputStream fos = new FileOutputStream(zip);
+ ZipOutputStream zos = new ZipOutputStream(fos)) {
+ for (ZipContent entry : entries) {
+ zos.putNextEntry(new ZipEntry(entry.name()));
+ zos.write(entry.payload().getBytes(StandardCharsets.UTF_8));
+ zos.closeEntry();
+ }
+ }
+ return zip;
+ }
+
+ @Test
+ public void parserSkipsEmptyZipEntryAndParsesRemaining(@TempDir Path tmp)
throws Exception {
+ JSONObject dagRoot = new JSONObject().put("dag", minimalDagJson());
+
+ File zip = writeZip(tmp, "empty-then-good.zip",
+ new ZipContent("empty-part.json", ""),
+ new ZipContent("whitespace-part.json", " \n\t "),
+ new ZipContent(DAG_ID, dagRoot.toString()));
+
+ ATSFileParser parser = new ATSFileParser(Collections.singletonList(zip));
+ DagInfo info = parser.getDAGData(DAG_ID);
+
+ assertNotNull(info, "Parser should return DagInfo even when some entries
are empty");
+ assertEquals(DAG_ID, info.getDagId());
+ assertEquals("SUCCEEDED", info.getStatus());
+ }
+
+ @Test
+ public void parserReportsOffendingEntryOnMalformedJson(@TempDir Path tmp)
throws Exception {
+ // Simulates the timeline server returning an HTML error page instead of
JSON.
+ File zip = writeZip(tmp, "malformed.zip",
+ new ZipContent(DAG_ID, "<html><body>internal error</body></html>"));
+
+ ATSFileParser parser = new ATSFileParser(Collections.singletonList(zip));
+ TezException thrown = assertThrows(TezException.class, () ->
parser.getDAGData(DAG_ID));
+
+ Throwable cause = thrown.getCause();
+ assertNotNull(cause);
+ String msg = cause.getMessage();
+ assertNotNull(msg);
+ assertTrue(msg.contains(DAG_ID),
+ "Error should name the offending zip entry, got: " + msg);
+ assertTrue(msg.contains("<html>"),
+ "Error should include a snippet of the offending payload, got: " +
msg);
+ }
+
+ private record ZipContent(String name, String payload) {
+ }
Review Comment:
what is the record for?
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestHistoryParser.java:
##########
@@ -234,6 +234,77 @@ public void testParserWithSuccessfulJob() throws Exception
{
isDAGEqual(dagInfoFromATS, shDagInfo);
}
+ /**
+ * the ATS write path is async (AM event queue → timeline client →
+ * timeline server). Even after the DAG client reports the job complete,
timeline entities
+ * may still be in transit. Downloading too early produced empty zip entries
and a
+ * misleading JSONException: A JSONObject text must begin with '{'} at parse
time.
+ */
+ private DagInfo fetchDagInfoFromAtsWithRetry(String dagId, int maxAttempts,
+ long delayMs) throws Exception {
+ String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR,
+ "--yarnTimelineAddress=" + yarnTimelineAddress };
+ Exception lastError = null;
+ DagInfo lastPartial = null;
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+ try {
+ // Fresh download every attempt — ATSImportTool overwrites the zip.
+ int result = ATSImportTool.process(args);
+ assertEquals(0, result);
+ DagInfo info = getDagInfo(dagId);
+ if (isDagInfoComplete(info)) {
+ return info;
+ }
+ lastPartial = info;
+ } catch (Exception e) {
+ lastError = e;
+ }
+ if (attempt < maxAttempts) {
+ Thread.sleep(delayMs);
+ }
+ }
+ fail("Could not fetch a complete DagInfo for " + dagId + " after " +
maxAttempts
+ + " attempts (lastError=" + lastError + ", lastPartial="
+ + (lastPartial == null ? "null"
+ : "vertices=" + lastPartial.getVertices().size()
+ + ", tasks=" + (lastPartial.getVertices().isEmpty() ? 0
+ :
lastPartial.getVertices().iterator().next().getTasks().size()))
+ + ")");
+ return null;
+ }
+
+ private static boolean isDagInfoComplete(DagInfo info) {
+ return info != null
+ && info.getVertices().size() >= 2
+ && info.getVertices().stream().allMatch(v ->
+ !v.getTasks().isEmpty()
+ && v.getTasks().stream().allMatch(t ->
!t.getTaskAttempts().isEmpty()));
+ }
+
+ private void waitForHistoryFileReady(String dagId, long timeoutMs) throws
Exception {
+ TezDAGID tezDAGID = TezDAGID.fromString(dagId);
+ ApplicationAttemptId applicationAttemptId =
ApplicationAttemptId.newInstance(tezDAGID
+ .getApplicationId(), 1);
Review Comment:
this can also fit into a single line
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestHistoryParser.java:
##########
@@ -234,6 +234,77 @@ public void testParserWithSuccessfulJob() throws Exception
{
isDAGEqual(dagInfoFromATS, shDagInfo);
}
+ /**
+ * the ATS write path is async (AM event queue → timeline client →
+ * timeline server). Even after the DAG client reports the job complete,
timeline entities
+ * may still be in transit. Downloading too early produced empty zip entries
and a
+ * misleading JSONException: A JSONObject text must begin with '{'} at parse
time.
+ */
+ private DagInfo fetchDagInfoFromAtsWithRetry(String dagId, int maxAttempts,
+ long delayMs) throws Exception {
+ String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR,
+ "--yarnTimelineAddress=" + yarnTimelineAddress };
+ Exception lastError = null;
+ DagInfo lastPartial = null;
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+ try {
+ // Fresh download every attempt — ATSImportTool overwrites the zip.
+ int result = ATSImportTool.process(args);
+ assertEquals(0, result);
+ DagInfo info = getDagInfo(dagId);
+ if (isDagInfoComplete(info)) {
+ return info;
+ }
+ lastPartial = info;
+ } catch (Exception e) {
+ lastError = e;
+ }
+ if (attempt < maxAttempts) {
+ Thread.sleep(delayMs);
+ }
+ }
+ fail("Could not fetch a complete DagInfo for " + dagId + " after " +
maxAttempts
+ + " attempts (lastError=" + lastError + ", lastPartial="
+ + (lastPartial == null ? "null"
+ : "vertices=" + lastPartial.getVertices().size()
+ + ", tasks=" + (lastPartial.getVertices().isEmpty() ? 0
+ :
lastPartial.getVertices().iterator().next().getTasks().size()))
+ + ")");
+ return null;
+ }
+
+ private static boolean isDagInfoComplete(DagInfo info) {
+ return info != null
+ && info.getVertices().size() >= 2
+ && info.getVertices().stream().allMatch(v ->
+ !v.getTasks().isEmpty()
+ && v.getTasks().stream().allMatch(t ->
!t.getTaskAttempts().isEmpty()));
+ }
+
+ private void waitForHistoryFileReady(String dagId, long timeoutMs) throws
Exception {
+ TezDAGID tezDAGID = TezDAGID.fromString(dagId);
+ ApplicationAttemptId applicationAttemptId =
ApplicationAttemptId.newInstance(tezDAGID
+ .getApplicationId(), 1);
+ Path historyPath = new Path(conf.get("fs.defaultFS")
+ + SIMPLE_HISTORY_DIR + HISTORY_TXT + "."
+ + applicationAttemptId);
+ FileSystem hfs = historyPath.getFileSystem(conf);
+ long deadline = System.currentTimeMillis() + timeoutMs;
Review Comment:
use monothonic time
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/parser/TestATSFileParser.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.tez.history.parser;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import org.apache.tez.dag.api.TezException;
+import org.apache.tez.history.parser.datamodel.DagInfo;
+
+import org.codehaus.jettison.json.JSONArray;
+import org.codehaus.jettison.json.JSONException;
+import org.codehaus.jettison.json.JSONObject;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class TestATSFileParser {
+
+ private static final String DAG_ID = "dag_1234567890_0001_1";
+
+ private static JSONObject minimalDagJson() throws JSONException {
+ JSONObject dag = new JSONObject();
+ dag.put("entityId", DAG_ID);
+ dag.put("entityType", "TEZ_DAG_ID");
+ JSONObject otherInfo = new JSONObject();
+ otherInfo.put("startTime", 1L);
+ otherInfo.put("endTime", 2L);
+ otherInfo.put("status", "SUCCEEDED");
+ otherInfo.put("counters", new JSONObject().put("counterGroups", new
JSONArray()));
+ dag.put("otherInfo", otherInfo);
+ return dag;
+ }
+
+ private static File writeZip(Path dir, String name, ZipContent... entries)
throws Exception {
+ File zip = dir.resolve(name).toFile();
+ try (FileOutputStream fos = new FileOutputStream(zip);
+ ZipOutputStream zos = new ZipOutputStream(fos)) {
+ for (ZipContent entry : entries) {
+ zos.putNextEntry(new ZipEntry(entry.name()));
+ zos.write(entry.payload().getBytes(StandardCharsets.UTF_8));
+ zos.closeEntry();
+ }
+ }
+ return zip;
+ }
+
+ @Test
+ public void parserSkipsEmptyZipEntryAndParsesRemaining(@TempDir Path tmp)
throws Exception {
+ JSONObject dagRoot = new JSONObject().put("dag", minimalDagJson());
+
+ File zip = writeZip(tmp, "empty-then-good.zip",
+ new ZipContent("empty-part.json", ""),
+ new ZipContent("whitespace-part.json", " \n\t "),
+ new ZipContent(DAG_ID, dagRoot.toString()));
+
+ ATSFileParser parser = new ATSFileParser(Collections.singletonList(zip));
+ DagInfo info = parser.getDAGData(DAG_ID);
+
+ assertNotNull(info, "Parser should return DagInfo even when some entries
are empty");
+ assertEquals(DAG_ID, info.getDagId());
+ assertEquals("SUCCEEDED", info.getStatus());
+ }
+
+ @Test
+ public void parserReportsOffendingEntryOnMalformedJson(@TempDir Path tmp)
throws Exception {
+ // Simulates the timeline server returning an HTML error page instead of
JSON.
+ File zip = writeZip(tmp, "malformed.zip",
+ new ZipContent(DAG_ID, "<html><body>internal error</body></html>"));
Review Comment:
no need to break line here
##########
tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/ATSFileParser.java:
##########
@@ -174,11 +175,23 @@ private void processApplication(JSONObject
tezApplicationJson) throws JSONExcept
}
}
- private JSONObject readJson(InputStream in) throws IOException,
JSONException {
- //Read entire content to memory
- final NonSyncByteArrayOutputStream bout = new
NonSyncByteArrayOutputStream();
- IOUtils.copy(in, bout);
- return new JSONObject(new String(bout.toByteArray(), "UTF-8"));
+ /**
+ * Parse the raw payload of a single zip entry as JSON.
+ * Returns null if the payload is empty or blank — callers should skip such
entries.
Review Comment:
empty or blank? what's the difference in this sense?
also, if there is a javadoc, it should state that a JSONException is thrown
in case of parse errors
##########
tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/ATSFileParser.java:
##########
@@ -197,7 +210,13 @@ private void parseATSZipFile(File atsFile)
ZipEntry zipEntry = zipEntries.nextElement();
LOG.debug("Processing " + zipEntry.getName());
InputStream inputStream = atsZipFile.getInputStream(zipEntry);
- JSONObject jsonObject = readJson(inputStream);
+ //Read entire content to memory so we can pass entry name into error
messages
+ final NonSyncByteArrayOutputStream bout = new
NonSyncByteArrayOutputStream();
+ IOUtils.copy(inputStream, bout);
+ JSONObject jsonObject = readJson(bout.toByteArray(),
zipEntry.getName());
+ if (jsonObject == null) {
+ continue;
+ }
Review Comment:
I think the all the mess with `NonSyncByteArrayOutputStream` can still go to
the `readJson`, right?
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/parser/TestATSFileParser.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.tez.history.parser;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import org.apache.tez.dag.api.TezException;
+import org.apache.tez.history.parser.datamodel.DagInfo;
+
+import org.codehaus.jettison.json.JSONArray;
+import org.codehaus.jettison.json.JSONException;
+import org.codehaus.jettison.json.JSONObject;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class TestATSFileParser {
+
+ private static final String DAG_ID = "dag_1234567890_0001_1";
+
+ private static JSONObject minimalDagJson() throws JSONException {
+ JSONObject dag = new JSONObject();
+ dag.put("entityId", DAG_ID);
+ dag.put("entityType", "TEZ_DAG_ID");
+ JSONObject otherInfo = new JSONObject();
+ otherInfo.put("startTime", 1L);
+ otherInfo.put("endTime", 2L);
+ otherInfo.put("status", "SUCCEEDED");
+ otherInfo.put("counters", new JSONObject().put("counterGroups", new
JSONArray()));
+ dag.put("otherInfo", otherInfo);
+ return dag;
+ }
+
+ private static File writeZip(Path dir, String name, ZipContent... entries)
throws Exception {
+ File zip = dir.resolve(name).toFile();
+ try (FileOutputStream fos = new FileOutputStream(zip);
+ ZipOutputStream zos = new ZipOutputStream(fos)) {
+ for (ZipContent entry : entries) {
+ zos.putNextEntry(new ZipEntry(entry.name()));
+ zos.write(entry.payload().getBytes(StandardCharsets.UTF_8));
+ zos.closeEntry();
+ }
+ }
+ return zip;
+ }
+
+ @Test
+ public void parserSkipsEmptyZipEntryAndParsesRemaining(@TempDir Path tmp)
throws Exception {
+ JSONObject dagRoot = new JSONObject().put("dag", minimalDagJson());
+
+ File zip = writeZip(tmp, "empty-then-good.zip",
+ new ZipContent("empty-part.json", ""),
+ new ZipContent("whitespace-part.json", " \n\t "),
+ new ZipContent(DAG_ID, dagRoot.toString()));
+
+ ATSFileParser parser = new ATSFileParser(Collections.singletonList(zip));
+ DagInfo info = parser.getDAGData(DAG_ID);
+
+ assertNotNull(info, "Parser should return DagInfo even when some entries
are empty");
+ assertEquals(DAG_ID, info.getDagId());
+ assertEquals("SUCCEEDED", info.getStatus());
+ }
+
+ @Test
+ public void parserReportsOffendingEntryOnMalformedJson(@TempDir Path tmp)
throws Exception {
+ // Simulates the timeline server returning an HTML error page instead of
JSON.
+ File zip = writeZip(tmp, "malformed.zip",
+ new ZipContent(DAG_ID, "<html><body>internal error</body></html>"));
+
+ ATSFileParser parser = new ATSFileParser(Collections.singletonList(zip));
+ TezException thrown = assertThrows(TezException.class, () ->
parser.getDAGData(DAG_ID));
+
+ Throwable cause = thrown.getCause();
+ assertNotNull(cause);
+ String msg = cause.getMessage();
+ assertNotNull(msg);
+ assertTrue(msg.contains(DAG_ID),
+ "Error should name the offending zip entry, got: " + msg);
+ assertTrue(msg.contains("<html>"),
+ "Error should include a snippet of the offending payload, got: " +
msg);
Review Comment:
no need for the line breaks I believe
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
