-
Notifications
You must be signed in to change notification settings - Fork 443
TEZ-4733: Fix flaky TestHistoryParser.testParserWithSuccessfulJob #519
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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; | ||
|
|
@@ -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()); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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 |
||
| } | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -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
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think the all the mess with |
||
|
|
||
| //This json can contain dag, vertices, tasks, task_attempts | ||
| JSONObject dagJson = jsonObject.optJSONObject(Constants.DAG); | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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. | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this is not needed: "Before TEZ-4733 that produced a misleading |
||
| 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); | ||
|
|
@@ -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++) { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. it's usually rather |
||
| 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); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think it's overkill to provide |
||
| } | ||
| } | ||
| 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
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. why is this check needed? is there a chance that the |
||
| } | ||
|
|
||
| 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
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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; | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. use monothonic time |
||
| long lastLen = -1L; | ||
| while (System.currentTimeMillis() < deadline) { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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 | ||
|
|
||
| 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
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. what is the record for? |
||
| } | ||
There was a problem hiding this comment.
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