TEZ-4733: Fix flaky TestHistoryParser.testParserWithSuccessfulJob - #519
Open
maheshrajus wants to merge 1 commit into
Open
TEZ-4733: Fix flaky TestHistoryParser.testParserWithSuccessfulJob#519maheshrajus wants to merge 1 commit into
maheshrajus wants to merge 1 commit into
Conversation
|
💔 -1 overall
This message was automatically generated. |
|
🎊 +1 overall
This message was automatically generated. |
Contributor
Author
|
@abstractdog Could you please review and approve the PR at your convenience? |
abstractdog
requested changes
Jul 29, 2026
| //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.
this is not needed: "Before TEZ-4733 that produced a misleading
//"A JSONObject text must begin with '{'" JSONException at parse time."
| lastError = e; | ||
| } | ||
| if (attempt < maxAttempts) { | ||
| Thread.sleep(delayMs); |
Contributor
There was a problem hiding this 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
Comment on lines
+277
to
+281
| return info != null | ||
| && info.getVertices().size() >= 2 | ||
| && info.getVertices().stream().allMatch(v -> | ||
| !v.getTasks().isEmpty() | ||
| && v.getTasks().stream().allMatch(t -> !t.getTaskAttempts().isEmpty())); |
Contributor
There was a problem hiding this 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
| FileSystem hfs = historyPath.getFileSystem(conf); | ||
| long deadline = System.currentTimeMillis() + timeoutMs; | ||
| long lastLen = -1L; | ||
| while (System.currentTimeMillis() < deadline) { |
| + SIMPLE_HISTORY_DIR + HISTORY_TXT + "." | ||
| + applicationAttemptId); | ||
| FileSystem hfs = historyPath.getFileSystem(conf); | ||
| long deadline = System.currentTimeMillis() + timeoutMs; |
Comment on lines
+103
to
+106
| 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); |
Contributor
There was a problem hiding this comment.
no need for the line breaks I believe
Comment on lines
+213
to
+219
| //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; | ||
| } |
Contributor
There was a problem hiding this comment.
I think the all the mess with NonSyncByteArrayOutputStream can still go to the readJson, right?
| 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. |
Contributor
There was a problem hiding this 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
| } 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.
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 "
Comment on lines
+109
to
+110
| private record ZipContent(String name, String payload) { | ||
| } |
Contributor
There was a problem hiding this comment.
what is the record for?
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
org.apache.tez.history.TestHistoryParser.testParserWithSuccessfulJob fails intermittently with:
Root cause
After the DAG client returns, ATSHistoryLoggingService still has history events in an async queue that must be
flushed to the timeline server. The test previously called ATSImportTool.process(...) immediately, so under load the
download could race the timeline write path — producing a zip whose entries were empty/whitespace, which then failed JSON parsing with the misleading "must begin with {" error.
A fixed Thread.sleep(10000) was already present before the SimpleHistory parse path (as a workaround for the same class of race), but there was no equivalent guard for the ATS parse path.
Changes
- skip empty/whitespace zip entries with a WARN; enrich JSON parse errors with the offending entry name + payload snippet.
-replaced the unguarded ATS export+parse with a retry loop that only accepts a DagInfo with ≥2
vertices, tasks, and attempts.
-replaced the fixed Thread.sleep(10000) before SimpleHistory parse with a
poll-until-file-size-stable.
-New tests added that checks (empty entry skipped, malformed entry names itself in the error).