@Override
  public String fromBytes(byte[] bytes) {
    Map<String, Object> values = messageSerde.fromBytes(bytes);
    CoordinatorStreamMessage message = new CoordinatorStreamMessage(new Object[]{}, values);
    if (type.equalsIgnoreCase(SetExecutionEnvContainerIdMapping.TYPE)) {
      SetExecutionEnvContainerIdMapping executionContainerIdMapping = new SetExecutionEnvContainerIdMapping(message);
      return executionContainerIdMapping.getExecutionEnvironmentContainerId();
    } else if (type.equalsIgnoreCase(SetContainerHostMapping.TYPE)) {
      SetContainerHostMapping hostMapping = new SetContainerHostMapping(message);
      return hostMapping.getHostLocality();
    } else if (type.equalsIgnoreCase(SetTaskContainerMapping.TYPE)) {
      SetTaskContainerMapping setTaskContainerMapping = new SetTaskContainerMapping(message);
      return setTaskContainerMapping.getTaskAssignment();
    } else if (type.equalsIgnoreCase(SetChangelogMapping.TYPE)) {
      SetChangelogMapping changelogMapping = new SetChangelogMapping(message);
      return (changelogMapping.getPartition() != null) ? String.valueOf(changelogMapping.getPartition()) : null;
    } else if (type.equalsIgnoreCase(SetConfig.TYPE)) {
      SetConfig setConfig = new SetConfig(message);
      return setConfig.getConfigValue();
    } else if (type.equalsIgnoreCase(SetTaskModeMapping.TYPE)) {
      SetTaskModeMapping setTaskModeMapping = new SetTaskModeMapping(message);
      return String.valueOf(setTaskModeMapping.getTaskMode());
    } else if (type.equalsIgnoreCase(SetTaskPartitionMapping.TYPE)) {
      SetTaskPartitionMapping setTaskPartitionMapping = new SetTaskPartitionMapping(message);
      return setTaskPartitionMapping.getTaskNames();
    } else if (type.equalsIgnoreCase(SetJobCoordinatorMetadataMessage.TYPE)) {
      SetJobCoordinatorMetadataMessage jobCoordinatorMetadataMessage = new SetJobCoordinatorMetadataMessage(message);
      return jobCoordinatorMetadataMessage.getJobCoordinatorMetadata();
    } else {
      throw new SamzaException(String.format("Unknown coordinator stream message type: %s", type));
    }
  }
Serializer for values written into the coordinator stream.
samza-core/src/main/java/org/apache/samza/coordinator/stream/CoordinatorStreamValueSerde.java