KafkaCheckpointLogKeySerde implements Serde<KafkaCheckpointLogKey> { @Override public byte[] toBytes(KafkaCheckpointLogKey key) { try { return MAPPER.writeValueAsBytes(ImmutableMap.of( SSP_GROUPER_FACTORY_FIELD, key.getGrouperFactoryClassName(), TASK_NAME_FIELD, key.getTaskName().toString(), TYPE_FIELD, key.getType() )); } catch (Exception e) { throw new SamzaException(String.format("Exception in serializing: %s", key), e); } }  @Override byte[] toBytes(KafkaCheckpointLogKey key); @Override KafkaCheckpointLogKey fromBytes(byte[] bytes);  }
@Test public void testBinaryCompatibility() { KafkaCheckpointLogKey logKey1 = new KafkaCheckpointLogKey(KafkaCheckpointLogKey.CHECKPOINT_KEY_TYPE, new TaskName("Partition 0"), GroupByPartitionFactory.class.getCanonicalName()); KafkaCheckpointLogKeySerde checkpointSerde = new KafkaCheckpointLogKeySerde(); byte[] bytes = ("{\"systemstreampartition-grouper-factory\"" + ":\"org.apache.samza.container.grouper.stream.GroupByPartitionFactory\",\"taskName\":\"Partition 0\"," + "\"type\":\"checkpoint\"}").getBytes(); Assert.assertEquals(true, Arrays.equals(bytes, checkpointSerde.toBytes(logKey1))); }