@SuppressWarnings("unchecked")
    // visible for testing
    public StitchRequestBody createStitchRequestBody(final Message inMessage) {
        if (inMessage.getBody() instanceof StitchRequestBody) {
            return createStitchRequestBodyFromStitchRequestBody(inMessage.getBody(StitchRequestBody.class), inMessage);
        }

        if (inMessage.getBody() instanceof StitchMessage) {
            return createStitchRequestBodyFromStitchMessages(Collections.singletonList(inMessage.getBody(StitchMessage.class)),
                    inMessage);
        }

        if (inMessage.getBody() instanceof Iterable) {
            return createStitchRequestBodyFromIterable(inMessage.getBody(Iterable.class), inMessage);
        }

        if (inMessage.getBody() instanceof Map) {
            return createStitchRecordFromMap(inMessage.getBody(Map.class), inMessage);
        }

        throw new IllegalArgumentException("Message body data `" + inMessage.getBody() + "` type is not supported");
    }
@Test
    void testIfCreateIfStitchMessagesSet() {
        final StitchConfiguration configuration = new StitchConfiguration();
        configuration.setTableName("test_table");
        configuration.setStitchSchema(StitchSchema.builder().addKeyword("field_1", "integer").build());
        configuration.setKeyNames("field_1,field_2");

        final StitchMessage message = StitchMessage.builder()
                .withData("field_1", "data")
                .withSequence(0)
                .build();

        final Exchange exchange = new DefaultExchange(context);
        exchange.getMessage().setBody(message);

        final StitchProducerOperations operations = new StitchProducerOperations(new TestClient(), configuration);

        assertEquals("{\"table_name\":\"test_table\",\"schema\":{\"field_1\":\"integer\"},\"messages\":[{\"action\":\"upsert\","
                     + "\"sequence\":0,\"data\":{\"field_1\":\"data\"}}],\"key_names\":[\"field_1\",\"field_2\"]}",
                JsonUtils.convertMapToJson(operations.createStitchRequestBody(exchange.getMessage()).toMap()));

        final StitchMessage message1 = StitchMessage.builder()
                .withData("field_1", "test_2")
                .withSequence(0)
                .build();

        exchange.getMessage().setHeader(StitchConstants.SCHEMA,
                StitchSchema.builder().addKeyword("field_1", "integer").addKeyword("field_2", "string").build());
        exchange.getMessage().setHeader(StitchConstants.TABLE_NAME, "test_table_2");
        exchange.getMessage().setHeader(StitchConstants.KEY_NAMES, "field_1,field_2");

        exchange.getMessage().setBody(message1);

        assertEquals("{\"table_name\":\"test_table_2\",\"schema\":{\"field_1\":\"integer\",\"field_2\":\"string\"},"
                     + "\"messages\":[{\"action\":\"upsert\",\"sequence\":0,\"data\":{\"field_1\":\"test_2\"}}],\"key_names\":[\"field_1\",\"field_2\"]}",
                JsonUtils.convertMapToJson(operations.createStitchRequestBody(exchange.getMessage()).toMap()));
    }