@Override
    public void serialize(ContainerPlacementMessage value, JsonGenerator jsonGenerator, SerializerProvider provider)
        throws IOException {
      Map<String, Object> containerPlacementMessageMap = new HashMap<String, Object>();
      if (value instanceof ContainerPlacementRequestMessage) {
        containerPlacementMessageMap.put("subType", ContainerPlacementRequestMessage.class.getSimpleName());
      } else if (value instanceof ContainerPlacementResponseMessage) {
        containerPlacementMessageMap.put("subType", ContainerPlacementResponseMessage.class.getSimpleName());
        containerPlacementMessageMap.put("responseMessage",
            ((ContainerPlacementResponseMessage) value).getResponseMessage());
      }
      if (value.getRequestExpiry() != null) {
        containerPlacementMessageMap.put("requestExpiry", value.getRequestExpiry().toMillis());
      }
      containerPlacementMessageMap.put("uuid", value.getUuid().toString());
      containerPlacementMessageMap.put("deploymentId", value.getDeploymentId());
      containerPlacementMessageMap.put("processorId", value.getProcessorId());
      containerPlacementMessageMap.put("destinationHost", value.getDestinationHost());
      containerPlacementMessageMap.put("statusCode", value.getStatusCode().name());
      containerPlacementMessageMap.put("timestamp", value.getTimestamp());
      jsonGenerator.writeObject(containerPlacementMessageMap);
    }
Object mapper for serializing and deserializing Container placement messages
samza-core/src/main/java/org/apache/samza/clustermanager/container/placement/ContainerPlacementMessageObjectMapper.java