@Test public void testMerge() throws InvalidProtocolBufferException { long receivedTime = System.currentTimeMillis(); double longitude = 10d; double latitude = 10d; SdiiMessage.Message sdiiMessageOne = prepareSDIIMessage(receivedTime, longitude, latitude); SdiiMessage.Message sdiiMessageTwo = prepareSDIIMessage(receivedTime, longitude + 0.01d, latitude + 0.01d); SdiiMessage.Message sdiiMessageThree = prepareSDIIMessage(receivedTime, longitude + 0.01d, latitude - 0.01d); SdiiMessage.Message sdiiMessageFour = prepareSDIIMessage(receivedTime, longitude - 0.01d, latitude - 0.01d); byte[] messageOne = sdiiMessageOne.toByteArray(); byte[] messageTwo = sdiiMessageTwo.toByteArray(); byte[] messageThree = sdiiMessageThree.toByteArray(); byte[] messageFour = sdiiMessageFour.toByteArray(); List<byte[]> messagesGroupOne = Arrays.asList(messageOne, messageTwo); Iterator<byte[]> messagesGroupOneIterator = messagesGroupOne.iterator(); List<byte[]> messagesGroupTwo = Arrays.asList(messageThree, messageFour); Iterator<byte[]> messagesGroupTwoIterator = messagesGroupTwo.iterator(); byte[] aggregatedMessagesGroupOne = aggregate(messagesGroupOneIterator); byte[] aggregatedMessagesGroupTwo = aggregate(messagesGroupTwoIterator); Iterator<byte[]> aggregatedMessagesIterator = Arrays.asList(aggregatedMessagesGroupOne, aggregatedMessagesGroupTwo).iterator(); ProtobufCompactionExample protobufCompactionExample = new ProtobufCompactionExample(); byte[] mergedMessages = protobufCompactionExample.merge(new HashMap<>(), aggregatedMessagesIterator); List<SdiiMessage.Message> mergedSdiiMessages = SdiiMessageList.MessageList.parseFrom(mergedMessages).getMessageList(); assertEquals(sdiiMessageOne, mergedSdiiMessages.get(0)); assertEquals(sdiiMessageTwo, mergedSdiiMessages.get(1)); assertEquals(sdiiMessageThree, mergedSdiiMessages.get(2)); assertEquals(sdiiMessageFour, mergedSdiiMessages.get(3)); }
@Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { List<SdiiMessage.Message> sdiiMessages = new ArrayList<>(); while (files.hasNext()) { byte[] file = files.next(); try { SdiiMessageList.MessageList sdiiMessageList = SdiiMessageList.MessageList.parseFrom(file); sdiiMessages.addAll(sdiiMessageList.getMessageList()); } catch (InvalidProtocolBufferException e) { LOG.error("Error merging sdii messages (key={})", keys.values(), e); return null; } } return SdiiMessageList.MessageList.newBuilder() .addAllMessage(sdiiMessages) .build() .toByteArray(); }
ProtobufCompactionExample implements CompactionUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { List<SdiiMessage.Message> sdiiMessages = new ArrayList<>(); while (files.hasNext()) { byte[] file = files.next(); try { SdiiMessageList.MessageList sdiiMessageList = SdiiMessageList.MessageList.parseFrom(file); sdiiMessages.addAll(sdiiMessageList.getMessageList()); } catch (InvalidProtocolBufferException e) { LOG.error("Error merging sdii messages (key={})", keys.values(), e); return null; } } return SdiiMessageList.MessageList.newBuilder() .addAllMessage(sdiiMessages) .build() .toByteArray(); } }
ProtobufCompactionExample implements CompactionUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { List<SdiiMessage.Message> sdiiMessages = new ArrayList<>(); while (files.hasNext()) { byte[] file = files.next(); try { SdiiMessageList.MessageList sdiiMessageList = SdiiMessageList.MessageList.parseFrom(file); sdiiMessages.addAll(sdiiMessageList.getMessageList()); } catch (InvalidProtocolBufferException e) { LOG.error("Error merging sdii messages (key={})", keys.values(), e); return null; } } return SdiiMessageList.MessageList.newBuilder() .addAllMessage(sdiiMessages) .build() .toByteArray(); }  }
ProtobufCompactionExample implements CompactionUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { List<SdiiMessage.Message> sdiiMessages = new ArrayList<>(); while (files.hasNext()) { byte[] file = files.next(); try { SdiiMessageList.MessageList sdiiMessageList = SdiiMessageList.MessageList.parseFrom(file); sdiiMessages.addAll(sdiiMessageList.getMessageList()); } catch (InvalidProtocolBufferException e) { LOG.error("Error merging sdii messages (key={})", keys.values(), e); return null; } } return SdiiMessageList.MessageList.newBuilder() .addAllMessage(sdiiMessages) .build() .toByteArray(); }  @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files); }
ProtobufCompactionExample implements CompactionUDF { @Override public byte[] merge(Map<String, Object> keys, Iterator<byte[]> files) { List<SdiiMessage.Message> sdiiMessages = new ArrayList<>(); while (files.hasNext()) { byte[] file = files.next(); try { SdiiMessageList.MessageList sdiiMessageList = SdiiMessageList.MessageList.parseFrom(file); sdiiMessages.addAll(sdiiMessageList.getMessageList()); } catch (InvalidProtocolBufferException e) { LOG.error("Error merging sdii messages (key={})", keys.values(), e); return null; } } return SdiiMessageList.MessageList.newBuilder() .addAllMessage(sdiiMessages) .build() .toByteArray(); }  @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files);  }