我是新的 Camel Kafka组件。我使用camel-kafka组件从kafka服务器发送和接收消息。我使用类似的代码，如下所述：

from("kafka:localhost:9092?topic=test&groupId=testing&autoOffsetReset=earliest&consumersCount=1")
                        .process(new Processor() {
                            @Override
                            public void process(Exchange exchange)
                                    throws Exception {
                                String messageKey = "";
                                if (exchange.getIn() != null) {
                                    Message message = exchange.getIn();
                                    Integer partitionId = (Integer) message
                                            .getHeader(KafkaConstants.PARTITION);
                                    String topicName = (String) message
                                            .getHeader(KafkaConstants.TOPIC);
                                    if (message.getHeader(KafkaConstants.KEY) != null)
                                        messageKey = (String) message
                                                .getHeader(KafkaConstants.KEY);
                                    Object data = message.getBody();
                                    System.out.println("topicName :: "
                                            + topicName + " partitionId :: "
                                            + partitionId + " messageKey :: "
                                            + messageKey + " message :: "
                                            + data + "\n");
   /// I perform many other operations here like persist the object in DB etc. 
                                }
                            }
                        });

这里的问题是，因为我没有发送任何确认回Kafka它是从服务器收到相同的消息三次。我的问题是，我怎样才能手动将应答发送回Kafka？我在camel kafka组件中没有找到任何合适的文档。