@Override
    public void transform(Message message, DataType fromType, DataType toType) {
        if (message.getHeaders().containsKey(Ddb2Constants.ITEM) ||
                message.getHeaders().containsKey(Ddb2Constants.KEY)) {
            return;
        }

        JsonNode jsonBody = getBodyAsJsonNode(message);

        String operation
                = Optional.ofNullable(jsonBody.get("operation")).map(JsonNode::asText).orElse(Ddb2Operations.PutItem.name());
        if (message.getExchange().hasProperties() && message.getExchange().getProperty("operation", String.class) != null) {
            operation = message.getExchange().getProperty("operation", String.class);
        }

        if (message.getHeaders().containsKey(Ddb2Constants.OPERATION)) {
            operation = message.getHeader(Ddb2Constants.OPERATION, Ddb2Operations.class).name();
        }

        JsonNode key = jsonBody.get("key");
        JsonNode item = jsonBody.get("item");

        Map<String, Object> keyProps;
        if (key != null) {
            keyProps = dataFormat.getObjectMapper().convertValue(key, new TypeReference<>() {
            });
        } else {
            keyProps = dataFormat.getObjectMapper().convertValue(jsonBody, new TypeReference<>() {
            });
        }

        Map<String, Object> itemProps;
        if (item != null) {
            itemProps = dataFormat.getObjectMapper().convertValue(item, new TypeReference<>() {
            });
        } else {
            itemProps = keyProps;
        }

        final Map<String, AttributeValue> keyMap = getAttributeValueMap(keyProps);

        switch (Ddb2Operations.valueOf(operation)) {
            case PutItem:
                message.setHeader(Ddb2Constants.OPERATION, Ddb2Operations.PutItem);
                message.setHeader(Ddb2Constants.ITEM, getAttributeValueMap(itemProps));
                setHeaderIfNotPresent(Ddb2Constants.RETURN_VALUES, ReturnValue.ALL_OLD.toString(), message);
                break;
            case UpdateItem:
                message.setHeader(Ddb2Constants.OPERATION, Ddb2Operations.UpdateItem);
                message.setHeader(Ddb2Constants.KEY, keyMap);
                message.setHeader(Ddb2Constants.UPDATE_VALUES, getAttributeValueUpdateMap(itemProps));
                setHeaderIfNotPresent(Ddb2Constants.RETURN_VALUES, ReturnValue.ALL_NEW.toString(), message);
                break;
            case DeleteItem:
                message.setHeader(Ddb2Constants.OPERATION, Ddb2Operations.DeleteItem);
                message.setHeader(Ddb2Constants.KEY, keyMap);
                setHeaderIfNotPresent(Ddb2Constants.RETURN_VALUES, ReturnValue.ALL_OLD.toString(), message);
                break;
            default:
                throw new UnsupportedOperationException(String.format("Unsupported operation '%s'", operation));
        }
    }
@Test
    @SuppressWarnings("unchecked")
    void shouldMapEmptyJson() throws Exception {
        Exchange exchange = new DefaultExchange(camelContext);

        exchange.getMessage().setBody("{}");
        exchange.getMessage().setHeader(Ddb2Constants.OPERATION, Ddb2Operations.PutItem.name());

        transformer.transform(exchange.getMessage(), DataType.ANY, new DataType(AWS_2_DDB_APPLICATION_JSON_TRANSFORMER));

        Assertions.assertTrue(exchange.getMessage().hasHeaders());
        Assertions.assertEquals(Ddb2Operations.PutItem, exchange.getMessage().getHeader(Ddb2Constants.OPERATION));
        Assertions.assertEquals(ReturnValue.ALL_OLD.toString(), exchange.getMessage().getHeader(Ddb2Constants.RETURN_VALUES));

        Map<String, AttributeValue> attributeValueMap = exchange.getMessage().getHeader(Ddb2Constants.ITEM, Map.class);
        Assertions.assertEquals(0L, attributeValueMap.size());
    }