@Override
    // Camel calls this method if the endpoint isSynchronous(), as the
    // KafkaEndpoint creates a SynchronousDelegateProducer for it
    public void process(Exchange exchange) throws Exception {
        // is the message body a list or something that contains multiple values
        Message message = exchange.getIn();

        if (transactionId != null) {
            startKafkaTransaction(exchange);
        }

        if (endpoint.getConfiguration().isUseIterator() && isIterable(message.getBody())) {
            processIterableSync(exchange, message);
        } else {
            processSingleMessageSync(exchange, message);
        }
    }
@Test
    public void processSendsMessageWithListOfExchangesWithIndividualHeaders() throws Exception {
        endpoint.getConfiguration().setBatchWithIndividualHeaders(true);
        endpoint.getConfiguration().setTopic("someTopic");
        Mockito.when(exchange.getIn()).thenReturn(in);
        Mockito.when(exchange.getMessage()).thenReturn(in);

        // we set the initial topic
        in.setHeader(KafkaConstants.OVERRIDE_TOPIC, "anotherTopic");
        in.setHeader(KafkaConstants.KEY, "someKey");
        in.setHeader(SOME_INDIVIDUAL_HEADER, "default");

        // we add our exchanges in order to aggregate
        final List<Exchange> nestedExchanges
                = createListOfExchangesWithTopics(Arrays.asList("overridenTopic1", "overridenTopic2", "overridenTopic3"));

        // aggregate
        final Exchange finalAggregatedExchange = aggregateExchanges(nestedExchanges, new GroupedExchangeAggregationStrategy());

        in.setBody(finalAggregatedExchange.getIn().getBody());
        in.setHeaders(finalAggregatedExchange.getIn().getHeaders());

        producer.process(exchange);

        // assert results
        verifySendMessages(Arrays.asList("overridenTopic1", "overridenTopic2", "overridenTopic3"),
                Arrays.asList("value-1", "value-2", "value-3"));
        assertRecordMetadataExists(3);
        assertRecordMetadataExistsForEachAggregatedExchange();
    }