@Override
    public ContainerPlacementMessage deserialize(JsonParser jsonParser, DeserializationContext context)
        throws IOException {
      ObjectCodec oc = jsonParser.getCodec();
      JsonNode node = oc.readTree(jsonParser);
      String subType = node.get("subType").textValue();
      String deploymentId = node.get("deploymentId").textValue();
      String processorId = node.get("processorId").textValue();
      String destinationHost = node.get("destinationHost").textValue();
      long timestamp = node.get("timestamp").longValue();
      Duration requestExpiry =
          node.get("requestExpiry") == null ? null : Duration.ofMillis(node.get("requestExpiry").longValue());
      ContainerPlacementMessage.StatusCode statusCode = null;
      UUID uuid = UUID.fromString(node.get("uuid").textValue());
      for (ContainerPlacementMessage.StatusCode code : ContainerPlacementMessage.StatusCode.values()) {
        if (code.name().equals(node.get("statusCode").textValue())) {
          statusCode = code;
        }
      }
      ContainerPlacementMessage message = null;
      if (subType.equals(ContainerPlacementRequestMessage.class.getSimpleName())) {
        message = new ContainerPlacementRequestMessage(uuid, deploymentId, processorId, destinationHost, requestExpiry, timestamp);
      } else if (subType.equals(ContainerPlacementResponseMessage.class.getSimpleName())) {
        String responseMessage = node.get("responseMessage").textValue();
        message = new ContainerPlacementResponseMessage(uuid, deploymentId, processorId, destinationHost, requestExpiry,
            statusCode, responseMessage, timestamp);
      }
      return message;
    }
Object mapper for serializing and deserializing Container placement messages
samza-core/src/main/java/org/apache/samza/clustermanager/container/placement/ContainerPlacementMessageObjectMapper.java