ProtoConcatenator { public static Message.Builder concatToProtobuf(long timestampMillis, long kafkaOffset, Collection<Message> messages) { try { final DynamicMessage.Builder messageBuilder = concatInner(messages, keys -> { try { return buildMessageBuilder("GeneratedObject", keys); } catch (Descriptors.DescriptorValidationException e) { LOGGER.error("Couldn't build concatenated Protobuf", e); throw new IllegalArgumentException(e); } }, (entry, builder) -> { String fieldName = entry.getKey().getName(); Descriptors.Descriptor descriptorForType = builder.getDescriptorForType(); Descriptors.FieldDescriptor dstFieldDescriptor = descriptorForType.findFieldByName(fieldName); if (dstFieldDescriptor == null) { throw new IllegalArgumentException("Tried to fill a non-existing field: " + fieldName); } if (dstFieldDescriptor.isRepeated()) { setRepeatedField(builder, dstFieldDescriptor, entry); } else { builder.setField(dstFieldDescriptor, entry.getValue()); } }); messageBuilder.setField(messageBuilder.getDescriptorForType().findFieldByName(TIMESTAMP_FIELD_NAME), timestampMillis); messageBuilder.setField(messageBuilder.getDescriptorForType().findFieldByName(KAFKA_OFFSET), kafkaOffset); return messageBuilder; } catch (IllegalArgumentException e) { LOGGER.error("Could not flatten Protobuf event", e); return null; } } protected  ProtoConcatenator(); static Message.Builder concatToProtobuf(long timestampMillis, long kafkaOffset, Collection<Message> messages); static Map<String, Object> concatToMap(long timestampMillis, Collection<Message> messages, boolean includeDefaultValueFields); static DynamicMessage.Builder buildMessageBuilder(String msgName, Collection<Descriptors.FieldDescriptor> fields); static final String TIMESTAMP_FIELD_NAME; }
@Test public void concatenateMessageWithItself() throws Descriptors.DescriptorValidationException { long kafkaOffset = 0L; DynamicMessage.Builder headerMessageBuilder = createHeaderMessageBuilder(); Descriptors.Descriptor headerMsgDesc = headerMessageBuilder.getDescriptorForType(); headerMessageBuilder.setField(headerMsgDesc.findFieldByName("id"), 1) .setField(headerMsgDesc.findFieldByName("name"), "one"); Map<String, Object> expectedValues = new HashMap<>(); expectedValues.put("id", 1); expectedValues.put("name", "one"); expectedValues.put(ProtoConcatenator.TIMESTAMP_FIELD_NAME, 0L); expectedValues.put(ProtoConcatenator.KAFKA_OFFSET, kafkaOffset); Assert.assertNull(ProtoConcatenator.concatToProtobuf(0, 0, Arrays.asList(headerMessageBuilder.build(), headerMessageBuilder.build()))); }