return callback.doInKafka(getTheProducer());

}</BUG>
@Override
<BUG>public void flush() {
getTheProducer().flush();

}</BUG>
protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> producerRecord) {
<BUG>getTheProducer();
if (this.logger.isTraceEnabled()) {</BUG>
this.logger.trace("Sending: " + producerRecord);

public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
protected final Log logger = LogFactory.getLog(this.getClass()); //NOSONAR
private final ProducerFactory<K, V> producerFactory;
private final boolean autoFlush;
private RecordMessageConverter messageConverter = new MessagingMessageConverter();
private volatile String defaultTopic;
private volatile ProducerListener<K, V> producerListener = new LoggingProducerListener<K, V>();
public KafkaTemplate(ProducerFactory<K, V> producerFactory) {
this(producerFactory, false);
