### Question:
AvroSimpleKeyExample implements SimpleUDF { @Override public byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages) { long startTime = System.currentTimeMillis(); try { Iterator<SdiiMessage.Message> sdiiMessages = StreamSupport.stream( Spliterators.spliteratorUnknownSize(messages, Spliterator.SIZED), true) .map( item -> { try { return SdiiMessage.Message.parseFrom(item); } catch (InvalidProtocolBufferException e) { throw new IllegalStateException(e); } }) .iterator(); return AvroHelper.aggregateProtobufMessagesAsAvro(sdiiMessages, SdiiMessage.Message.class); } catch (Exception e) { LOG.error("Aggregation errors....", e); } finally { aggregateDuration.inc(System.currentTimeMillis() - startTime); } return null; }  @Override void open(Configuration parameters, RuntimeContext runtimeContext); @Override Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload); @Override byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages); @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files);  }### Answer:
@Test public void testAggregate() throws IOException { SdiiMessage.Message message1 = prepareSDIIMessage(System.currentTimeMillis(), true, 10d, 10d); SdiiMessage.Message message2 = prepareSDIIMessage(System.currentTimeMillis(), true, 10d, 10d); SdiiMessage.Message message3 = prepareSDIIMessage(System.currentTimeMillis(), true, 10d, 10d); List<byte[]> messagesList = Arrays.asList(message1.toByteArray(), message2.toByteArray(), message3.toByteArray()); example.open(parameters, runtimeContext); assertEquals(0, aggregateDuration.getCount()); File tmpFile = File.createTempFile("test", ".avro"); tmpFile.deleteOnExit(); FileUtils.writeByteArrayToFile( tmpFile, example.aggregate(new HashMap<>(), messagesList.iterator())); List<SdiiMessage.Message> list = AvroHelper.fromFile(tmpFile, SdiiMessage.Message.class); assertEquals(3, list.size()); assertSDIIMessagesAreEqual(message1, list.get(0)); assertSDIIMessagesAreEqual(message2, list.get(1)); assertSDIIMessagesAreEqual(message3, list.get(2)); assertTrue(aggregateDuration.getCount() > 0); }