OutputDescriptor outputDescriptor = mock(OutputDescriptor.class);
when(outputDescriptor.getClassName()).thenReturn("OutputDescriptor");
OutputContext outputContext = new TezOutputContextImpl(conf, new String[] {workDir.toString()},
appAttemptNumber, tezUmbilical, dagName, taskVertexName, destinationVertexName,
-1, taskAttemptID, counters, 0, userPayload, runtimeTask,
null, auxEnv, new MemoryDistributor(1, 1, conf) , outputDescriptor, null);
UnorderedKVOutput kvOutput = new OnFileUnorderedKVOutputForTest(outputContext, 1);
List<Event> events = null;
events = kvOutput.initialize();
assertTrue(events != null && events.size() == 0);
KeyValueWriter kvWriter = kvOutput.getWriter();
List<KVPair> data = KVDataGen.generateTestData(true);
for (KVPair kvp : data) {
kvWriter.write(kvp.getKey(), kvp.getvalue());
}
events = kvOutput.close();
assertTrue(events != null && events.size() == 1);
DataMovementEvent dmEvent = (DataMovementEvent)events.get(0);
assertEquals("Invalid source index", 0, dmEvent.getSourceIndex());
DataMovementEventPayloadProto shufflePayload = DataMovementEventPayloadProto
.parseFrom(ByteString.copyFrom(dmEvent.getUserPayload()));
assertFalse(shufflePayload.hasEmptyPartitions());
assertEquals(outputContext.getUniqueIdentifier(), shufflePayload.getPathComponent());
assertEquals(shufflePort, shufflePayload.getPort());
assertEquals("host", shufflePayload.getHost());
}