### Question:
ProtobufSimpleKeyExample implements SimpleUDF { @Override public byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages) { List<SdiiMessage.Message> aggregatedList = new ArrayList<>(); while (messages.hasNext()) { byte[] data = messages.next(); try { SdiiMessage.Message m = SdiiMessage.Message.parseFrom(data); aggregatedList.add(m); } catch (InvalidProtocolBufferException e) { LOG.error("Aggregation errors....", e); return null; } } return SdiiMessageList.MessageList.newBuilder() .addAllMessage(aggregatedList) .build() .toByteArray(); }  @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 InvalidProtocolBufferException { List<byte[]> list = Arrays.asList(b1, b2); Iterator it = list.iterator(); byte[] aggregatedByteArray = example.aggregate(new HashMap<>(), it); List<SdiiMessage.Message> aggregatedMessageList = SdiiMessageList.MessageList.parseFrom(aggregatedByteArray).getMessageList(); assertEquals(m1, aggregatedMessageList.get(0)); assertEquals(m2, aggregatedMessageList.get(1)); }