Kafka Sink Connector（对接Elasticsearch）工作机制与offset提交咨询

Great question—building a Kafka Sink Connector for Elasticsearch means getting a solid grasp of how Connect’s Worker interacts with your SinkTask, especially around offsets and flushes. Let’s break this down clearly:

1. Core SinkTask Workflow Overview #

First, let’s set the stage for how the Connect Worker runs your task:

• The Worker runs an infinite loop to process Kafka records:
  • It polls Kafka Brokers for batches of records (controlled by configs like consumer.max.poll.records)
  • It passes these records to your put(List<SinkRecord>) method
  • Based on timing or buffer triggers, it calls flush(Map<TopicPartition, OffsetAndMetadata>)
  • Finally, it handles offset commits (tied directly to the success of your flush logic)

2. Offset Commit Timing & Its Link to flush() #

Offset commits are how Connect tracks your task’s progress to avoid reprocessing records. The timing is tightly coupled to the flush() method, with two key scenarios:

Default Auto-Flush + Commit #

By default, Connect uses offset.flush.interval.ms (default 60000ms = 1 minute) to trigger flushes. Here’s what happens:

• When the interval elapses, the Worker calls your flush() method, passing in the highest offsets it’s sent to put()
• The Worker waits for your flush() to complete successfully (respecting offset.flush.timeout.ms for timeouts)
• Only after flush() returns without errors will the Worker commit those offsets to Kafka’s internal __consumer_offsets topic
• If flush() throws an exception, the Worker retries the flush (using retry.backoff.ms) and holds off on committing offsets until it succeeds

Manual/Buffer-Based Flush (Recommended for Reliability) #

For production use cases (like guaranteeing records are written to Elasticsearch before committing offsets), you’ll want to buffer records in put() and only flush when ready:

• In put(), add records to an in-memory buffer instead of sending them directly to Elasticsearch (this reduces overhead from repeated HTTP calls)
• Trigger an early flush in put() if your buffer hits a configured batch size (e.g., 1000 records)
• When the Worker calls flush() (either on interval or rebalance), write all buffered records to Elasticsearch via bulk requests
• Only return from flush() once you’ve confirmed all records are successfully indexed (handle retries for failed bulk requests here)
• The Worker will commit the offsets passed to flush() immediately after your method succeeds

3. Critical Technical Details You Might Be Missing #

Let’s cover some often-overlooked nuances that matter for a robust Elasticsearch sink:

• Offset Storage: Connect stores offsets in Kafka’s __consumer_offsets topic, not Elasticsearch. This means your task’s progress is safe even if Elasticsearch goes down.
• Rebalance Handling: When a task is rebalanced (e.g., scaling the connector), the Worker will first call flush() to commit pending offsets, then close() your task. This prevents data loss during rebalancing.
• Exactly-Once Semantics: To guarantee no duplicate records in Elasticsearch:
  • Use Elasticsearch’s bulk API with idempotent requests (set op_type: create or use unique document IDs)
  • Enable transactional.id in your connector config to leverage Kafka’s transactional capabilities
  • Ensure flush() only returns after all bulk requests are confirmed successful
• Error Handling: If some records fail to index:
  • Throw an exception to trigger flush retries (configure max.retries and retry.backoff.ms)
  • Use a dead-letter queue (DLQ) by setting errors.deadletterqueue.topic.name to send failed records to a separate Kafka topic, allowing successful records to be flushed and offsets committed
• start() & close() Best Practices:
  • Use start() to initialize your Elasticsearch client, load configs (endpoints, credentials), and set up bulk processors
  • In close(), perform a final flush of any remaining buffered records, then clean up the Elasticsearch client and release resources

Example Snippet for Context #

Here’s a simplified version of how your SinkTask might implement this logic:

public class ElasticsearchSinkTask extends SinkTask {
    private RestHighLevelClient esClient;
    private List<SinkRecord> recordBuffer;
    private int batchSize;

    @Override
    public void start(Map<String, String> props) {
        // Initialize Elasticsearch client
        esClient = new RestHighLevelClient(RestClient.builder(new HttpHost("localhost", 9200)));
        batchSize = Integer.parseInt(props.getOrDefault("batch.size", "1000"));
        recordBuffer = new ArrayList<>(batchSize);
    }

    @Override
    public void put(List<SinkRecord> records) {
        recordBuffer.addAll(records);
        // Flush early if buffer reaches batch size
        if (recordBuffer.size() >= batchSize) {
            flushBufferToElasticsearch();
        }
    }

    @Override
    public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // Ensure all buffered records are written before committing offsets
        flushBufferToElasticsearch();
    }

    private void flushBufferToElasticsearch() {
        if (recordBuffer.isEmpty()) return;

        BulkRequest bulkRequest = new BulkRequest();
        for (SinkRecord record : recordBuffer) {
            // Convert SinkRecord to Elasticsearch IndexRequest
            IndexRequest indexReq = new IndexRequest("my-target-index")
                    .id(record.key().toString())
                    .source(record.value(), XContentType.JSON);
            bulkRequest.add(indexReq);
        }

        try {
            BulkResponse bulkResponse = esClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulkResponse.hasFailures()) {
                // Throw exception to trigger retry, or route failures to DLQ
                throw new RuntimeException("Bulk index failed: " + bulkResponse.buildFailureMessage());
            }
            recordBuffer.clear();
        } catch (IOException e) {
            throw new RuntimeException("Failed to send bulk request to Elasticsearch", e);
        }
    }

    @Override
    public void close() {
        try {
            // Final flush before shutting down
            flushBufferToElasticsearch();
            esClient.close();
        } catch (IOException e) {
            log.error("Error closing Elasticsearch client", e);
        }
    }

    @Override
    public String version() {
        return "1.0.0";
    }
}

内容的提问来源于stack exchange，提问作者Dogil