Protobuf data corruption when CloudEvent is serialized/deserialized several times #523

    @Test
    public void testMultipleSerializationDeserialization() throws InvalidProtocolBufferException {
        EventFormat serializer = new ProtobufFormat();

        Status status = Status.newBuilder().setCode(Code.UNKNOWN_VALUE).build();
        CloudEvent event1 = buildCloudEvent("id123", "test", "http:/test", status);
        CloudEventData data1 = event1.getData();
        assertTrue(data1 instanceof ProtoCloudEventData);
        Message message1 = ((ProtoCloudEventData) data1).getMessage();
        //assertTrue(message1 instanceof Any); // Not Any???
        assertTrue(message1 instanceof Status);
        assertEquals(status, message1);

        byte[] raw1 = serializer.serialize(event1);
        System.out.println("  event1: " + event1);
        System.out.println("    raw1: " + Arrays.toString(raw1));
        System.out.println("   size1: " + raw1.length);
        System.out.println("payload1: " + message1);

        CloudEvent event2 = serializer.deserialize(raw1);
        CloudEventData data2 = event2.getData();
        assertTrue(data2 instanceof ProtoCloudEventData);
        Message message2 = ((ProtoCloudEventData) data2).getMessage();
        assertTrue(message2 instanceof Any);
        assertEquals(status, ((Any) message2).unpack(Status.class));
        byte[] raw2 = serializer.serialize(event2);
        System.out.println("  event2: " + event2);
        System.out.println("    raw2: " + Arrays.toString(raw2));
        System.out.println("   size2: " + raw2.length);
        System.out.println("payload2: " + message2);
        //assertEquals(event1, event2);
        assertArrayEquals(raw1, raw2);

        CloudEvent event3 = serializer.deserialize(raw2);
        CloudEventData data3 = event3.getData();
        assertTrue(data3 instanceof ProtoCloudEventData);
        Message message3 = ((ProtoCloudEventData) data3).getMessage();
        assertTrue(message3 instanceof Any);
        assertEquals(status, (((Any) message3).unpack(Any.class).unpack(Status.class)));
        byte[] raw3 = serializer.serialize(event3);
        System.out.println("  event3: " + event3);
        System.out.println("    raw3: " + Arrays.toString(raw3));
        System.out.println("   size3: " + raw3.length);
        System.out.println("payload3: " + message3);
        //assertEquals(event1, event3);
        //assertEquals(event2, event3);
        assertArrayEquals(raw1, raw3);
        assertArrayEquals(raw2, raw3);
    }

    static CloudEvent buildCloudEvent(String id, String type, String source, Message message) {
        return CloudEventBuilder.v1()
                .withId(id)
                .withType(type)
                .withSource(URI.create(source))
                .withData(ProtoCloudEventData.wrap(message))
                .build();
    }