@Override
  public byte[] toBytes(String value) {
    if (type.equalsIgnoreCase(SetExecutionEnvContainerIdMapping.TYPE)) {
      SetExecutionEnvContainerIdMapping
          executionEnvContainerIdMapping = new SetExecutionEnvContainerIdMapping(SOURCE, "", value);
      return messageSerde.toBytes(executionEnvContainerIdMapping.getMessageMap());
    } else if (type.equalsIgnoreCase(SetContainerHostMapping.TYPE)) {
      SetContainerHostMapping hostMapping = new SetContainerHostMapping(SOURCE, "", value, "", "");
      return messageSerde.toBytes(hostMapping.getMessageMap());
    } else if (type.equalsIgnoreCase(SetTaskContainerMapping.TYPE)) {
      SetTaskContainerMapping setTaskContainerMapping = new SetTaskContainerMapping(SOURCE, "", value);
      return messageSerde.toBytes(setTaskContainerMapping.getMessageMap());
    } else if (type.equalsIgnoreCase(SetTaskModeMapping.TYPE)) {
      SetTaskModeMapping setTaskModeMapping = new SetTaskModeMapping(SOURCE, "", value);
      return messageSerde.toBytes(setTaskModeMapping.getMessageMap());
    } else if (type.equalsIgnoreCase(SetChangelogMapping.TYPE)) {
      SetChangelogMapping changelogMapping = new SetChangelogMapping(SOURCE, "", Integer.valueOf(value));
      return messageSerde.toBytes(changelogMapping.getMessageMap());
    } else if (type.equalsIgnoreCase(SetConfig.TYPE)) {
      SetConfig setConfig = new SetConfig(SOURCE, "", value);
      return messageSerde.toBytes(setConfig.getMessageMap());
    } else if (type.equalsIgnoreCase(SetTaskPartitionMapping.TYPE)) {
      SetTaskPartitionMapping setTaskPartitionMapping = new SetTaskPartitionMapping(SOURCE, "", value);
      return messageSerde.toBytes(setTaskPartitionMapping.getMessageMap());
    } else if (type.equalsIgnoreCase(SetJobCoordinatorMetadataMessage.TYPE)) {
      SetJobCoordinatorMetadataMessage jobCoordinatorMetadataMessage = new SetJobCoordinatorMetadataMessage(SOURCE, "", value);
      return messageSerde.toBytes(jobCoordinatorMetadataMessage.getMessageMap());
    } 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