protected Consumer<K, V> createConsumer() {
        // Always force reset to the beginning of the log since this class wants to consume all available log data
        consumerConfigs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // Turn off autocommit since we always want to consume the full log
        consumerConfigs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        return new KafkaConsumer<>(consumerConfigs);
    }
Send a record asynchronously to the configured {@link #topic}.
@param key the key for the {@link ProducerRecord}
@param value the value for the {@link ProducerRecord}
@param callback the callback to invoke after completion; can be null if no callback is desired

@return the future from the call to {@link Producer#send}. {@link Future#get} can be called on this returned
        future if synchronous behavior is desired.
connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java