deniskuzZ commented on code in PR #540:
URL: https://github.com/apache/tez/pull/540#discussion_r4144654893
##########
tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleInputEventHandlerImpl.java:
##########
@@ -408,6 +408,63 @@ public void testDataMovementEventsWithShuffleData() throws
IOException {
}
}
+ /**
+ * An input whose partition is empty reports zero rows and must still reach
the ShuffleManager:
+ * counting it only in numInputs and not in the denominator is what inflated
the counter.
+ */
+ @Test
+ @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+ public void testAnEmptyPartitionStillReportsItsRecordCount() throws
IOException {
+ ShuffleManager shuffleManager = mock(ShuffleManager.class);
+ ShuffleInputEventHandlerImpl handler = newHandler(shuffleManager, false);
+
+ handler.handleEvents(Collections.singletonList(
+ createRecordCountEvent(1, 0, createEmptyPartitionByteString(0),
false)));
+
+ verify(shuffleManager).updateApproximateInputRecords(eq(1), eq(0L));
+ verify(shuffleManager).addCompletedInputWithNoData(any());
+ }
+
+ /** The composite path carries the same payload and has to report it too. */
+ @Test
+ @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+ public void testACompositeEventReportsItsRecordCount() throws IOException {
+ ShuffleManager shuffleManager = mock(ShuffleManager.class);
+ ShuffleInputEventHandlerImpl handler = newHandler(shuffleManager, true);
+
+ handler.handleEvents(Collections.singletonList(createRecordCountEvent(1,
4000, null, true)));
+
+ verify(shuffleManager).updateApproximateInputRecords(eq(1), eq(4000L));
+ }
+
+ /**
+ * A source task can emit more rows than an int32 holds. num_record is an
int64 so the count
+ * survives the wire; passing it as an int would not even compile against
that field.
+ */
+ @Test
+ @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+ public void testACountAboveIntMaxSurvivesTheWire() throws IOException {
+ ShuffleManager shuffleManager = mock(ShuffleManager.class);
+ ShuffleInputEventHandlerImpl handler = newHandler(shuffleManager, false);
+
+ handler.handleEvents(Collections.singletonList(
+ createRecordCountEvent(1, 3_000_000_000L, null, false)));
+
+ verify(shuffleManager).updateApproximateInputRecords(eq(1),
eq(3_000_000_000L));
+ }
+
+ /** A payload with no record count -- any multi-partition writer -- reports
nothing. */
+ @Test
+ @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+ public void testAPayloadWithoutARecordCountReportsNothing() throws
IOException {
+ ShuffleManager shuffleManager = mock(ShuffleManager.class);
+ ShuffleInputEventHandlerImpl handler = newHandler(shuffleManager, false);
+
+ handler.handleEvents(Collections.singletonList(createDataMovementEvent(0,
1, null)));
+
+ verify(shuffleManager, times(0)).updateApproximateInputRecords(anyInt(),
anyLong());
+ }
Review Comment:
fixed
--
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]
