public Exchange createDbzExchange(DebeziumConsumer consumer, final SourceRecord sourceRecord) {
        final Exchange exchange;
        if (consumer != null) {
            exchange = consumer.createExchange(false);
        } else {
            exchange = super.createExchange();
        }

        final Message message = exchange.getIn();

        final Schema valueSchema = sourceRecord.valueSchema();
        final Object value = sourceRecord.value();

        // extract values from SourceRecord
        final Map<String, Object> sourceMetadata = extractSourceMetadataValueFromValueStruct(valueSchema, value);
        final Object operation = extractValueFromValueStruct(valueSchema, value, Envelope.FieldName.OPERATION);
        final Object before = extractValueFromValueStruct(valueSchema, value, Envelope.FieldName.BEFORE);
        final Object body = extractBodyValueFromValueStruct(valueSchema, value);
        final Object timestamp = extractValueFromValueStruct(valueSchema, value, Envelope.FieldName.TIMESTAMP);
        final Object ddl = extractValueFromValueStruct(valueSchema, value, HistoryRecord.Fields.DDL_STATEMENTS);
        // set message headers
        message.setHeader(DebeziumConstants.HEADER_IDENTIFIER, sourceRecord.topic());
        message.setHeader(DebeziumConstants.HEADER_KEY, sourceRecord.key());
        message.setHeader(DebeziumConstants.HEADER_SOURCE_METADATA, sourceMetadata);
        message.setHeader(DebeziumConstants.HEADER_OPERATION, operation);
        message.setHeader(DebeziumConstants.HEADER_BEFORE, before);
        message.setHeader(DebeziumConstants.HEADER_TIMESTAMP, timestamp);
        message.setHeader(DebeziumConstants.HEADER_DDL_SQL, ddl);
        message.setHeader(Exchange.MESSAGE_TIMESTAMP, timestamp);

        message.setBody(body);

        return exchange;
    }
@Test
    void testIfCreatesExchangeFromSourceRecordOtherThanStruct() {
        final SourceRecord sourceRecord = createStringRecord();

        final Exchange exchange = debeziumEndpoint.createDbzExchange(null, sourceRecord);
        final Message inMessage = exchange.getIn();

        assertNotNull(exchange);

        // assert headers
        assertEquals("dummy", inMessage.getHeader(DebeziumConstants.HEADER_IDENTIFIER));
        assertNull(inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));

        // assert value
        final String value = (String) inMessage.getBody();
        assertEquals(sourceRecord.value(), value);
    }