Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.List;
Expand Down Expand Up @@ -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.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

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

*/
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());

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

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 "

}
}

/**
Expand All @@ -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;
}
Comment on lines +213 to +219

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I think the all the mess with NonSyncByteArrayOutputStream can still go to the readJson, right?


//This json can contain dag, vertices, tasks, task_attempts
JSONObject dagJson = jsonObject.optJSONObject(Constants.DAG);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

this is not needed: "Before TEZ-4733 that produced a misleading
//"A JSONObject text must begin with '{'" JSONException at parse time."

DagInfo dagInfoFromATS = fetchDagInfoFromAtsWithRetry(dagId, 6, 5_000L);

int result = ATSImportTool.process(args);
assertEquals(0, result);

//Parse ATS data and verify results
DagInfo dagInfoFromATS = getDagInfo(dagId);
verifyDagInfo(dagInfoFromATS, true);
verifyJobSpecificInfo(dagInfoFromATS);
checkConfig(dagInfoFromATS);

//Now run with SimpleHistoryLogging
dagId = runWordCount(WordCount.TokenProcessor.class.getName(),
WordCount.SumProcessor.class.getName(), "WordCount", false);
Thread.sleep(10000); //For all flushes to happen and to avoid half-cooked download.

waitForHistoryFileReady(dagId, 60_000L);

DagInfo shDagInfo = getDagInfoFromSimpleHistory(dagId);
verifyDagInfo(shDagInfo, false);
Expand All @@ -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++) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

it's usually rather int attempt = 0; attempt < maxAttempts;, isn't it?

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);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

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

}
}
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()));
Comment on lines +277 to +281

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

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

}

private void waitForHistoryFileReady(String dagId, long timeoutMs) throws Exception {
TezDAGID tezDAGID = TezDAGID.fromString(dagId);
ApplicationAttemptId applicationAttemptId = ApplicationAttemptId.newInstance(tezDAGID
.getApplicationId(), 1);
Comment on lines +286 to +287

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

this can also fit into a single line

Path historyPath = new Path(conf.get("fs.defaultFS")
+ SIMPLE_HISTORY_DIR + HISTORY_TXT + "."
+ applicationAttemptId);
Comment on lines +288 to +290

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

too many line breaks, it can be less I think

FileSystem hfs = historyPath.getFileSystem(conf);
long deadline = System.currentTimeMillis() + timeoutMs;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

use monothonic time

long lastLen = -1L;
while (System.currentTimeMillis() < deadline) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

use monothonic time

if (hfs.exists(historyPath)) {
long len = hfs.getFileStatus(historyPath).getLen();
if (len > 0 && len == lastLen) {
return;
}
lastLen = len;
}
Thread.sleep(500);
}
fail("Timed out waiting for SimpleHistory file " + historyPath
+ " to be ready within " + timeoutMs + "ms (lastLen=" + lastLen + ")");
}

private DagInfo getDagInfoFromSimpleHistory(String dagId) throws TezException, IOException {
TezDAGID tezDAGID = TezDAGID.fromString(dagId);
ApplicationAttemptId applicationAttemptId = ApplicationAttemptId.newInstance(tezDAGID
Expand Down
Original file line number Diff line number Diff line change
@@ -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>"));
Comment on lines +93 to +94

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

no need to break line here


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);
Comment on lines +103 to +106

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

no need for the line breaks I believe

}

private record ZipContent(String name, String payload) {
}
Comment on lines +109 to +110

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

what is the record for?

}