props.put(BridgeConfig.BOOTSTRAP_SERVERS, BridgeConfig.getBootstrapServers());
props.put(BridgeConfig.KEY_DESERIALIZER, BridgeConfig.getKeyDeserializer());
props.put(BridgeConfig.VALUE_DESERIALIZER, BridgeConfig.getValueDeserializer());
props.put(BridgeConfig.GROUP_ID, groupId);
props.put(BridgeConfig.AUTO_OFFSET_RESET, "earliest"); // TODO : it depends on x-opt-bridge.offset inside the AMQP message
<BUG>this.kafkaConsumerRunner = new KafkaConsumerRunner(props, topic, this.queue, this.vertx, this.ebQueue);
</BUG>
this.kafkaConsumerThread = new Thread(kafkaConsumerRunner);
this.kafkaConsumerThread.start();
this.ebConsumer = this.vertx.eventBus().consumer(this.ebQueue, ebMessage -> {

props.put(BridgeConfig.BOOTSTRAP_SERVERS, BridgeConfig.getBootstrapServers());
props.put(BridgeConfig.KEY_DESERIALIZER, BridgeConfig.getKeyDeserializer());
props.put(BridgeConfig.VALUE_DESERIALIZER, BridgeConfig.getValueDeserializer());
props.put(BridgeConfig.GROUP_ID, groupId);
props.put(BridgeConfig.AUTO_OFFSET_RESET, "earliest"); // TODO : it depends on x-opt-bridge.offset inside the AMQP message
this.kafkaConsumerRunner = new KafkaConsumerRunner(props, topic, this.queue, this.vertx, this.ebQueue, sender.getQoS(), this.barrier);
this.kafkaConsumerThread = new Thread(kafkaConsumerRunner);
this.kafkaConsumerThread.start();
this.ebConsumer = this.vertx.eventBus().consumer(this.ebQueue, ebMessage -> {
