Copilot commented on code in PR #519:
URL: https://github.com/apache/tez/pull/519#discussion_r3766678908
##########
tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/ATSFileParser.java:
##########
@@ -190,14 +210,18 @@ private JSONObject readJson(InputStream in) throws
IOException, JSONException {
*/
private void parseATSZipFile(File atsFile)
throws IOException, JSONException, TezException, InterruptedException {
- final ZipFile atsZipFile = new ZipFile(atsFile);
- try {
+ try (ZipFile atsZipFile = new ZipFile(atsFile)) {
Enumeration<? extends ZipEntry> zipEntries = atsZipFile.entries();
while (zipEntries.hasMoreElements()) {
ZipEntry zipEntry = zipEntries.nextElement();
LOG.debug("Processing " + zipEntry.getName());
- InputStream inputStream = atsZipFile.getInputStream(zipEntry);
- JSONObject jsonObject = readJson(inputStream);
+ JSONObject jsonObject;
+ try (InputStream inputStream = atsZipFile.getInputStream(zipEntry)) {
+ jsonObject = readJson(inputStream, zipEntry.getName());
+ }
+ if (jsonObject == null) {
+ continue;
Review Comment:
Skipping blanks leaves `dagInfo` null when the archive has no nonblank DAG
entry. `getDAGData` then unconditionally calls `addRawDataToDagInfo`, causing a
`NullPointerException` at `BaseParser.java:64` rather than the declared
`TezException`. After processing all entries, explicitly reject an archive that
did not initialize `dagInfo` with a diagnostic naming the archive/DAG.
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestHistoryParser.java:
##########
@@ -234,6 +235,83 @@ 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,
+ * so an export triggered right after completion can capture a partial
snapshot (empty zip
+ * entries, or a DagInfo with fewer vertices / empty task lists). Retry the
export+parse
+ * pipeline until {@link #isDagInfoComplete} confirms the snapshot is
populated.
+ */
+ private DagInfo fetchDagInfoFromAtsWithRetry(String dagId, int maxAttempts,
+ int expectedNumOfVertices) throws Exception {
+ String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR,
+ "--yarnTimelineAddress=" + yarnTimelineAddress };
+ Exception lastError = null;
+ DagInfo lastPartial = null;
+ for (int attempt = 0; attempt < maxAttempts; attempt++) {
+ try {
+ // Fresh download every attempt — ATSImportTool overwrites the zip.
+ int result = ATSImportTool.process(args);
+ assertEquals(0, result);
Review Comment:
`ATSImportTool.run` converts download failures to a `-1` result, but this
assertion throws `AssertionFailedError`, which is not caught by the `catch
(Exception)` below. A transient export failure therefore aborts on the first
attempt instead of being retried. Convert a nonzero result to an `Exception`
inside the retry block.
##########
tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestHistoryParser.java:
##########
@@ -234,6 +235,83 @@ 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,
+ * so an export triggered right after completion can capture a partial
snapshot (empty zip
+ * entries, or a DagInfo with fewer vertices / empty task lists). Retry the
export+parse
+ * pipeline until {@link #isDagInfoComplete} confirms the snapshot is
populated.
+ */
+ private DagInfo fetchDagInfoFromAtsWithRetry(String dagId, int maxAttempts,
+ int expectedNumOfVertices) throws Exception {
+ String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR,
+ "--yarnTimelineAddress=" + yarnTimelineAddress };
+ Exception lastError = null;
+ DagInfo lastPartial = null;
+ for (int attempt = 0; 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, expectedNumOfVertices)) {
+ return info;
+ }
+ lastPartial = info;
+ } catch (Exception e) {
+ lastError = e;
+ }
+ if (attempt < maxAttempts - 1) {
+ Thread.sleep(ATS_RETRY_DELAY_MS);
+ }
+ }
+ 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;
+ }
+
+ /**
+ * ATS parsing can succeed on a partially-written zip: the reader returns a
DagInfo with
+ * fewer vertices than the DAG actually has, or vertices whose task/attempt
lists are still
+ * empty. That's the race we're guarding against — a "successful" parse is
not proof the
+ * export was complete. We know the expected vertex count from the DAG under
test, so require
+ * it explicitly and require every vertex to have at least one task with at
least one attempt.
+ */
+ private static boolean isDagInfoComplete(DagInfo info, int
expectedNumOfVertices) {
+ return info != null
+ && info.getVertices().size() >= expectedNumOfVertices
+ && 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 deadlineNanos = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(timeoutMs);
+ long lastLen = -1L;
+ while (System.nanoTime() < deadlineNanos) {
+ if (hfs.exists(historyPath)) {
+ long len = hfs.getFileStatus(historyPath).getLen();
+ if (len > 0 && len == lastLen) {
+ return;
Review Comment:
Two equal length samples 500 ms apart do not establish that the async
history writer has finished; it can pause between events while the file is
still open. This can return before the terminal DAG event is appended and
reintroduce the partial-history parse this helper replaces. Poll an actual
completion signal, such as the HDFS file being closed or the terminal DAG
record being present, rather than short-term size stability.
--
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]
