EventDeserializerSchema extends AbstractDeserializationSchema<GenericEvent> { @Override public GenericEvent deserialize(byte[] message) { UntypedEvent event; try { event = objectMapper.readValue(message, UntypedEvent.class); } catch (IOException e) { logger.error("deserialization failed for object", e); return null; } return event.asEvent(); } EventDeserializerSchema(); @Override GenericEvent deserialize(byte[] message); @Override TypeInformation<GenericEvent> getProducedType();  }
@Test public void serAndDeser() { GenericEvent event = GenericEventBuilder.anEvent() .withEvent("Type") .withTimestamp(1L) .withSource("me") .withParameter("a", Value.of(1L)) .withParameter("b", Value.of("String")) .build(); GenericEvent event1; EventSerializationSchema ser = new EventSerializationSchema(); EventDeserializerSchema deSer = new EventDeserializerSchema(); byte[] bytes = ser.serialize(event); event1 = deSer.deserialize(bytes); Assert.assertEquals(event, event1); }