7.3 – Sink Connector Implementation

Overview

This module teaches how to implement Sink Connectors that push data from Kafka to external systems. By the end, you will understand:

• SinkTask mechanics and the put() method.
• Batch writes and buffering strategies.
• Delivery guarantees and idempotent writes.
• External system integration patterns.

This module focuses exclusively on Sink Connector development. Source Connector development was covered in Module 7.2.

Sink Connector Architecture

Data Flow

Kafka Topic
    │
    ▼
Internal Kafka Consumer ───► SinkTask.put(records)
    │
    ├─ Batch records
    ├─ Transform to external format
    │
    ▼
External System (e.g., Database, Elasticsearch, S3)
    │
    ▼
Offset Commit ───► __consumer_offsets topic

Key Responsibilities

1. Consume from Kafka: Internal consumer fetches records from assigned partitions.
2. Buffer Records: Accumulate records into batches for efficient writes.
3. Transform to External Format: Convert SinkRecord to external system format (e.g., SQL INSERT, JSON document).
4. Write to External System: Execute batch writes (e.g., database transaction, bulk API call).
5. Commit Offsets: Commit Kafka consumer offsets after successful writes.

The SinkTask API

Interface: org.apache.kafka.connect.sink.SinkTask

All sink tasks extend SinkTask and implement three key methods:

1. void start(Map<String, String> props): Initialize the task.
2. void put(Collection<SinkRecord> records): Process a batch of records from Kafka.
3. void stop(): Clean up resources.

Implementing start(Map<String, String> props)

Purpose

Initialize the task with task-specific configuration and connect to the external system.

Example: JDBC Sink Task

@Override
public void start(Map<String, String> props) {
    // Parse task configuration
    String connectionUrl = props.get("connection.url");
    String tableName = props.get("table.name");
    
    // Connect to database
    try {
        this.connection = DriverManager.getConnection(connectionUrl);
        this.connection.setAutoCommit(false); // Use transactions
    } catch (SQLException e) {
        throw new ConnectException("Failed to connect to database", e);
    }
    
    // Initialize prepared statement
    try {
        String sql = String.format(
            "INSERT INTO %s (id, name, email) VALUES (?, ?, ?) " +
            "ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email",
            tableName
        );
        this.preparedStatement = connection.prepareStatement(sql);
    } catch (SQLException e) {
        throw new ConnectException("Failed to prepare statement", e);
    }
}

Key Steps

1. Parse Configuration: Extract task-specific properties.
2. Connect to External System: Establish database connection, HTTP client, S3 client, etc.
3. Prepare Resources: Create prepared statements, initialize buffers, etc.

Implementing put(Collection<SinkRecord> records)

Purpose

Process a batch of records from Kafka and write them to the external system.

Signature

@Override
public void put(Collection<SinkRecord> records) {
    // Iterate over records
    // Write to external system
    // Commit transaction (if applicable)
}

Put Loop Behavior

Connect runtime repeatedly calls put():

1. Consumer fetches records from Kafka (batch size controlled by consumer.max.poll.records).
2. Connect calls put(records) with the fetched batch.
3. Task writes records to external system.
4. Connect commits Kafka consumer offsets (after put() completes without exception).
5. Repeat (loop continues until task is stopped).

Example: JDBC Sink Task

@Override
public void put(Collection<SinkRecord> records) {
    try {
        for (SinkRecord record : records) {
            // Extract value (assume Struct format)
            Struct value = (Struct) record.value();
            
            long id = value.getInt64("id");
            String name = value.getString("name");
            String email = value.getString("email");
            
            // Add to batch
            preparedStatement.setLong(1, id);
            preparedStatement.setString(2, name);
            preparedStatement.setString(3, email);
            preparedStatement.addBatch();
        }
        
        // Execute batch
        int[] results = preparedStatement.executeBatch();
        
        // Commit transaction
        connection.commit();
        
    } catch (SQLException e) {
        // Rollback on error
        try {
            connection.rollback();
        } catch (SQLException rollbackEx) {
            // Log rollback failure
        }
        throw new ConnectException("Failed to write to database", e);
    }
}

Key Steps

1. Iterate Records: Loop over SinkRecord objects in the batch.
2. Extract Data: Parse record key/value (may be Struct, Map, or primitive).
3. Buffer Writes: Add to batch (e.g., PreparedStatement.addBatch(), buffer list).
4. Execute Batch: Write all records in one operation (e.g., executeBatch(), bulk API call).
5. Commit: Commit transaction or finalize write.
6. Handle Errors: Rollback on failure, throw ConnectException to trigger retry/DLQ.

SinkRecord Anatomy

Key Fields

public class SinkRecord {
    String topic();               // Kafka topic name
    int kafkaPartition();         // Kafka partition
    long kafkaOffset();           // Kafka offset
    Schema keySchema();           // Key schema (nullable)
    Object key();                 // Key (nullable)
    Schema valueSchema();         // Value schema
    Object value();               // Value
    Long timestamp();             // Record timestamp (milliseconds)
    Map<String, ?> headers();     // Record headers
}

Example: Extract Value

Struct value = (Struct) record.value();
long userId = value.getInt64("user_id");
String userName = value.getString("name");

Example: Use Key for Primary Key

String key = (String) record.key();
// Use key as primary key in database

Schema and Struct Handling in Sink Connectors

In sink connectors, you'll often need to extract data from Schema + Struct objects (produced by source connectors or Kafka producers using structured formats) and convert them to the external system's format.

Why Schema and Struct?

Kafka Connect uses org.apache.kafka.connect.data.Schema and org.apache.kafka.connect.data.Struct to represent structured data. Sink tasks must:

1. Parse Schema: Understand field types and structure.
2. Extract Values: Retrieve field values from Struct.
3. Map to External System: Convert to database columns, JSON, etc.

Reading Schema and Struct

@Override
public void put(Collection<SinkRecord> records) {
    for (SinkRecord record : records) {
        // Get schema and value
        Schema valueSchema = record.valueSchema();
        Object valueObject = record.value();
        
        // Cast to Struct (if schema is present)
        if (valueSchema != null && valueObject instanceof Struct) {
            Struct value = (Struct) valueObject;
            
            // Extract fields
            Long id = value.getInt64("id");
            String name = value.getString("name");
            String email = value.getString("email");
            
            // Write to external system
            writeToDatabase(id, name, email);
        } else {
            // Handle schemaless data (e.g., Map<String, Object>)
            Map<String, Object> value = (Map<String, Object>) valueObject;
            // ...
        }
    }
}

Schema Types and Field Extraction

Example: Database Insert with Struct

@Override
public void put(Collection<SinkRecord> records) {
    try {
        PreparedStatement stmt = connection.prepareStatement(
            "INSERT INTO users (id, name, email, created_at) VALUES (?, ?, ?, ?)"
        );
        
        for (SinkRecord record : records) {
            Struct value = (Struct) record.value();
            
            // Extract fields with type safety
            Long id = value.getInt64("id");
            String name = value.getString("name");
            String email = value.getString("email");
            java.util.Date createdAt = (java.util.Date) value.get("created_at");
            
            // Set parameters
            stmt.setLong(1, id);
            stmt.setString(2, name);
            stmt.setString(3, email);
            stmt.setTimestamp(4, new java.sql.Timestamp(createdAt.getTime()));
            
            stmt.addBatch();
        }
        
        stmt.executeBatch();
        connection.commit();
        
    } catch (SQLException e) {
        throw new ConnectException("Failed to write to database", e);
    }
}

Handling Optional Fields

Scenario: Schema defines optional fields (e.g., email is nullable).

Schema valueSchema = SchemaBuilder.struct()
    .field("id", Schema.INT64_SCHEMA)
    .field("name", Schema.STRING_SCHEMA)
    .field("email", Schema.OPTIONAL_STRING_SCHEMA) // Optional
    .build();

Extraction:

Struct value = (Struct) record.value();

Long id = value.getInt64("id");
String name = value.getString("name");
String email = value.getString("email"); // May be null

// Handle null
if (email == null) {
    email = "unknown@example.com";
}

Dynamic Schema Handling

Problem: Schema changes at runtime (new fields added).

Solution: Iterate over schema fields dynamically:

Schema valueSchema = record.valueSchema();
Struct value = (Struct) record.value();

// Get all field names
for (Field field : valueSchema.fields()) {
    String fieldName = field.name();
    Schema.Type fieldType = field.schema().type();
    
    // Extract value based on type
    switch (fieldType) {
        case STRING:
            String stringValue = value.getString(fieldName);
            break;
        case INT64:
            Long longValue = value.getInt64(fieldName);
            break;
        // ... handle other types
    }
}

Use Case: Auto-evolving database schemas (e.g., auto.evolve=true in JDBC Sink).

Schemaless Data (JSON without Schema)

Scenario: Connector receives JSON data without schema (e.g., schemas.enable=false).

Example Record:

{"id": 123, "name": "Alice", "email": "alice@example.com"}

Extraction:

@Override
public void put(Collection<SinkRecord> records) {
    for (SinkRecord record : records) {
        // Value is a Map<String, Object>
        Map<String, Object> value = (Map<String, Object>) record.value();
        
        // Extract fields (with type casting)
        Long id = ((Number) value.get("id")).longValue();
        String name = (String) value.get("name");
        String email = (String) value.get("email");
        
        writeToDatabase(id, name, email);
    }
}

Trade-off: No type safety, requires manual casting and validation.

Schema-Based vs. Schemaless Comparison

Recommendation: Use Schema + Struct for production deployments to ensure data quality and evolution support.

Example: Nested Structs

Scenario: Record contains nested objects.

Schema:

Schema addressSchema = SchemaBuilder.struct()
    .field("street", Schema.STRING_SCHEMA)
    .field("city", Schema.STRING_SCHEMA)
    .field("zip", Schema.STRING_SCHEMA)
    .build();

Schema userSchema = SchemaBuilder.struct()
    .field("id", Schema.INT64_SCHEMA)
    .field("name", Schema.STRING_SCHEMA)
    .field("address", addressSchema) // Nested struct
    .build();

Extraction:

Struct value = (Struct) record.value();

Long id = value.getInt64("id");
String name = value.getString("name");

// Extract nested struct
Struct address = value.getStruct("address");
String street = address.getString("street");
String city = address.getString("city");
String zip = address.getString("zip");

Use Case: Complex data models (e.g., user profiles with nested addresses, order with line items).

Batch Writes and Buffering

Why Batch?

Single-record writes (e.g., one INSERT per record) are inefficient:

• High network overhead (many round-trips).
• Low throughput (database locks per write).

Batch writes (e.g., one bulk INSERT for 1000 records) are efficient:

• Fewer round-trips.
• Higher throughput (amortized locks).

Batching Strategies

1. Fixed-Size Batching

Strategy: Buffer N records, then write.

Example:

List<SinkRecord> buffer = new ArrayList<>();

@Override
public void put(Collection<SinkRecord> records) {
    buffer.addAll(records);
    
    if (buffer.size() >= batchSize) {
        writeBatch(buffer);
        buffer.clear();
    }
}

Pros: Simple, predictable memory usage.
Cons: Last batch may be incomplete if task stops.

2. Time-Based Batching

Strategy: Buffer for T seconds, then write.

Example (requires background thread):

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

@Override
public void start(Map<String, String> props) {
    scheduler.scheduleAtFixedRate(() -> {
        if (!buffer.isEmpty()) {
            writeBatch(buffer);
            buffer.clear();
        }
    }, flushIntervalMs, flushIntervalMs, TimeUnit.MILLISECONDS);
}

Pros: Lower latency (records aren't held indefinitely).
Cons: More complex (requires scheduler).

3. Hybrid Batching (Recommended)

Strategy: Write when buffer is full or every T seconds, whichever comes first.

Best Practice: Use Connect's built-in batching (controlled by consumer.max.poll.records) and write all records in put() at once.

Delivery Guarantees

At-Least-Once Delivery (Default)

By default, sink connectors provide at-least-once delivery:

1. Connect calls put(records).
2. Task writes records to external system.
3. Connect commits Kafka consumer offsets.

Failure Scenario: If the task crashes after writing but before offset commit, records will be re-consumed and re-written (duplicates).

Achieving Exactly-Once (Idempotency)

Strategy: Use idempotent writes in the external system.

Option 1: Upsert (Database)

Example (PostgreSQL):

INSERT INTO users (id, name, email) VALUES (?, ?, ?)
ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email

Result: Duplicate writes are harmless (last write wins).

Option 2: Unique IDs (Elasticsearch)

Example:

IndexRequest request = new IndexRequest("users")
    .id(String.valueOf(userId)) // Use user ID as document ID
    .source(jsonMap);

Result: Duplicate writes overwrite the same document (idempotent).

Option 3: Deduplication (Application-Level)

Strategy: Track processed offsets in the external system and skip duplicates.

Example (Database):

CREATE TABLE processed_offsets (
    topic VARCHAR,
    partition INT,
    offset BIGINT,
    PRIMARY KEY (topic, partition)
);

Before Writing:

SELECT offset FROM processed_offsets WHERE topic = ? AND partition = ?;

If offset > last processed, write and update processed_offsets.

External System Integration Patterns

Pattern 1: Batch Database Inserts (JDBC)

Use Case: Write records to a relational database.

Strategy: Use PreparedStatement.addBatch() for bulk inserts.

Example:

for (SinkRecord record : records) {
    Struct value = (Struct) record.value();
    preparedStatement.setLong(1, value.getInt64("id"));
    preparedStatement.setString(2, value.getString("name"));
    preparedStatement.addBatch();
}
preparedStatement.executeBatch();
connection.commit();

Best Practice: Use upsert mode for idempotency.

Pattern 2: Bulk Indexing (Elasticsearch)

Use Case: Index documents in Elasticsearch.

Strategy: Use Elasticsearch Bulk API.

Example:

BulkRequest bulkRequest = new BulkRequest();

for (SinkRecord record : records) {
    Struct value = (Struct) record.value();
    Map<String, Object> jsonMap = structToMap(value);
    
    IndexRequest indexRequest = new IndexRequest("users")
        .id(String.valueOf(value.getInt64("user_id")))
        .source(jsonMap);
    
    bulkRequest.add(indexRequest);
}

BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);

if (bulkResponse.hasFailures()) {
    throw new ConnectException("Bulk indexing failed: " + bulkResponse.buildFailureMessage());
}

Best Practice: Use document IDs for idempotency.

Pattern 3: File Writes (S3, HDFS)

Use Case: Write records to files in S3 or HDFS.

Strategy: Buffer records, write to local file, upload to S3 on rotation.

Example:

BufferedWriter writer = new BufferedWriter(new FileWriter("/tmp/output.json"));

for (SinkRecord record : records) {
    String json = recordToJson(record);
    writer.write(json);
    writer.newLine();
}

writer.close();

// Upload to S3
s3Client.putObject("my-bucket", "output-" + System.currentTimeMillis() + ".json", new File("/tmp/output.json"));

Best Practice: Use partition/offset in file naming for idempotency.

Pattern 4: REST API Writes

Use Case: POST records to a REST API.

Strategy: Batch records into JSON array, send bulk POST request.

Example:

JSONArray jsonArray = new JSONArray();

for (SinkRecord record : records) {
    Struct value = (Struct) record.value();
    JSONObject json = new JSONObject();
    json.put("id", value.getInt64("id"));
    json.put("name", value.getString("name"));
    jsonArray.put(json);
}

HttpPost request = new HttpPost("https://api.example.com/bulk");
request.setEntity(new StringEntity(jsonArray.toString()));
HttpResponse response = httpClient.execute(request);

if (response.getStatusLine().getStatusCode() != 200) {
    throw new ConnectException("API write failed: " + response.getStatusLine());
}

Best Practice: Use retries for transient failures (network timeouts).

Example: Elasticsearch Sink Connector

Scenario

Write user records from Kafka to Elasticsearch.

Implementation

public class ElasticsearchSinkTask extends SinkTask {
    private RestHighLevelClient client;
    private String indexName;

    @Override
    public void start(Map<String, String> props) {
        String host = props.get("connection.url");
        this.indexName = props.get("index.name");
        
        // Initialize Elasticsearch client
        this.client = new RestHighLevelClient(
            RestClient.builder(HttpHost.create(host))
        );
    }

    @Override
    public void put(Collection<SinkRecord> records) {
        BulkRequest bulkRequest = new BulkRequest();
        
        for (SinkRecord record : records) {
            Struct value = (Struct) record.value();
            
            Map<String, Object> document = new HashMap<>();
            document.put("id", value.getInt64("id"));
            document.put("name", value.getString("name"));
            document.put("email", value.getString("email"));
            
            IndexRequest indexRequest = new IndexRequest(indexName)
                .id(String.valueOf(value.getInt64("id")))
                .source(document);
            
            bulkRequest.add(indexRequest);
        }
        
        try {
            BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            
            if (bulkResponse.hasFailures()) {
                throw new ConnectException("Bulk indexing failed: " + bulkResponse.buildFailureMessage());
            }
        } catch (IOException e) {
            throw new ConnectException("Failed to index documents", e);
        }
    }

    @Override
    public void stop() {
        try {
            if (client != null) {
                client.close();
            }
        } catch (IOException e) {
            // Log error
        }
    }

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

Offset Commit Control (Advanced)

Default Behavior

Connect automatically commits offsets after put() completes without throwing an exception.

Custom Commit Logic

For advanced use cases, you can override preCommit() to control offset commits:

@Override
public Map<TopicPartition, OffsetAndMetadata> preCommit(
    Map<TopicPartition, OffsetAndMetadata> currentOffsets
) {
    // Custom logic: only commit if external system confirms write
    if (externalSystemConfirmsWrite()) {
        return currentOffsets; // Commit these offsets
    } else {
        return Collections.emptyMap(); // Don't commit (will retry)
    }
}

Use Case: Transactional writes where Kafka offsets must be committed atomically with external system writes.

Key Takeaways

1. SinkTask.put() is the core method—process batch of records, write to external system.
2. Batch writes are essential for performance (use bulk APIs, prepared statements).
3. At-least-once delivery by default; achieve exactly-once via idempotent writes (upsert, unique IDs).
4. Offset commits happen automatically after put() completes successfully.
5. External integration: JDBC batch inserts, Elasticsearch bulk API, S3 file writes, REST API posts.

What's Next?

• Module 7.4 explains testing and packaging custom connectors.
• Module 6.1 reviews error handling and retries (relevant for sink failures).
• Module 2.1 reviews Source vs. Sink semantics (cross-reference).

Further Reading

• SinkTask API Docs
• Sink Connector Tutorial (Confluent)

End of Module 7.3