deniskuzZ commented on code in PR #540:
URL: https://github.com/apache/tez/pull/540#discussion_r4144773686
##########
tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java:
##########
@@ -432,6 +433,72 @@ public void testTextMixedRecordsWithoutFinalMerge(boolean
shouldCompress, Report
textTest(100, 10, 2048, 10, 10, 10, false, false);
}
+ /**
+ * One partition with pipelined shuffle is the only shape where
writeLargeRecord runs while
+ * num_record is set, so it is the only one that can show the DME
under-counting large records.
+ */
+ @ParameterizedTest(name = "test[{0}, {1}]")
+ @MethodSource("data")
+ @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
+ public void testLargeRecords_SinglePartition(boolean shouldCompress,
+ ReportPartitionStats reportPartitionStats) throws IOException,
InterruptedException {
+ setupInit(shouldCompress, reportPartitionStats);
+ textTest(0, 1, 2048, 0, 0, 5, true, false);
+ }
+
+ /**
+ * An output small enough to fit one buffer spills only at close, so its
records reach the task
+ * counters in the same call that builds the final event -- which is where a
count read too
+ * early reports zero.
+ */
+ @ParameterizedTest(name = "test[{0}, {1}]")
+ @MethodSource("data")
+ @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
+ public void testFinalEventRecordCount_SinglePartition(boolean shouldCompress,
+ ReportPartitionStats reportPartitionStats) throws IOException,
InterruptedException {
+ setupInit(shouldCompress, reportPartitionStats);
+ textTest(5, 1, 2048, 0, 0, 0, true, false);
+ }
+
+ /**
+ * A single-partition writer is the only one that fills num_record, and
large records never
+ * reach outputRecordsCounter -- writeLargeRecord bypasses it -- so both the
DME and the
+ * VertexManager event have to carry the sum of the two counters. Under
pipelined shuffle the
+ * events are pushed through sendEvents rather than returned by close(), so
both sources are
+ * scanned and the last of each kind is the one that has to be right.
+ */
+ private void assertLastEventsCountEveryRecord(UnorderedPartitionedKVWriter
kvWriter,
+ OutputContext outputContext, List<Event> closeEvents) throws IOException
{
+ List<Event> allEvents = new ArrayList<>(closeEvents);
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<Event>> sent = ArgumentCaptor.forClass(List.class);
+ verify(outputContext, atLeast(0)).sendEvents(sent.capture());
+ sent.getAllValues().forEach(allEvents::addAll);
+
+ long expected = kvWriter.outputRecordsCounter.getValue()
+ + kvWriter.outputLargeRecordsCounter.getValue();
Review Comment:
changed
--
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]
