7.2 – Source Connector Implementation

Overview

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

• SourceTask mechanics and the poll() method.
• Offset storage and management.
• Poll model and loop structure.
• External system integration patterns.

This module focuses exclusively on Source Connector development. Sink Connector development is covered in Module 7.3.

Source Connector Architecture

Data Flow

External System
    │
    ├─ poll() ───► SourceTask.poll()
    │
    ├─ Transform to SourceRecord
    │
    ▼
Internal Kafka Producer ───► Kafka Topic
    │
    ▼
Offset Commit ───► connect-offsets topic

Key Responsibilities

1. Poll External System: Query external system for new data (e.g., database, file, API).
2. Convert to SourceRecord: Transform external data into Connect's SourceRecord format.
3. Produce to Kafka: Use internal producer to write records to Kafka.
4. Commit Offsets: Store source offsets (e.g., row ID, file position) for resumption.

The SourceTask API

Interface: org.apache.kafka.connect.source.SourceTask

All source tasks extend SourceTask and implement two key methods:

1. void start(Map<String, String> props): Initialize the task.
2. List<SourceRecord> poll(): Poll external system for new records.
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 Source Task

@Override
public void start(Map<String, String> props) {
    // Parse task configuration
    String tableName = props.get("table.name");
    String connectionUrl = props.get("connection.url");
    
    // Connect to database
    try {
        this.connection = DriverManager.getConnection(connectionUrl);
        this.statement = connection.createStatement();
    } catch (SQLException e) {
        throw new ConnectException("Failed to connect to database", e);
    }
    
    // Load last offset (if resuming)
    Map<String, Object> offset = context.offsetStorageReader()
        .offset(Collections.singletonMap("table", tableName));
    
    if (offset != null) {
        this.lastOffset = (Long) offset.get("id");
    } else {
        this.lastOffset = 0L; // Start from beginning
    }
}

Key Steps

1. Parse Configuration: Extract task-specific properties.
2. Connect to External System: Establish database connection, open file, create HTTP client, etc.
3. Load Last Offset: Use context.offsetStorageReader().offset(sourcePartition) to resume from where the task left off.

Implementing poll()

Purpose

Poll the external system for new records and return them as a list of SourceRecord objects.

Signature

@Override
public List<SourceRecord> poll() throws InterruptedException {
    // Query external system
    // Convert results to SourceRecords
    // Return list (or null if no data)
}

Poll Loop Behavior

Connect runtime repeatedly calls poll():

1. Task calls poll().
2. Task returns a list of SourceRecord objects (or null if no data).
3. Connect produces records to Kafka.
4. Connect commits offsets.
5. Repeat (loop continues until task is stopped).

Example: JDBC Source Task

@Override
public List<SourceRecord> poll() throws InterruptedException {
    List<SourceRecord> records = new ArrayList<>();
    
    try {
        // Query database for new rows
        String query = String.format(
            "SELECT * FROM %s WHERE id > %d ORDER BY id LIMIT 1000",
            tableName, lastOffset
        );
        
        ResultSet rs = statement.executeQuery(query);
        
        while (rs.next()) {
            long id = rs.getLong("id");
            String name = rs.getString("name");
            String email = rs.getString("email");
            
            // Create source partition (identifies logical partition)
            Map<String, String> sourcePartition = Collections.singletonMap("table", tableName);
            
            // Create source offset (position in external system)
            Map<String, Long> sourceOffset = Collections.singletonMap("id", id);
            
            // Create Kafka record value
            Struct value = new Struct(schema)
                .put("id", id)
                .put("name", name)
                .put("email", email);
            
            // Create SourceRecord
            SourceRecord record = new SourceRecord(
                sourcePartition,      // Source partition
                sourceOffset,         // Source offset
                topicName,            // Kafka topic
                Schema.STRING_SCHEMA, // Key schema
                String.valueOf(id),   // Key (user ID)
                schema,               // Value schema
                value                 // Value (user data)
            );
            
            records.add(record);
            lastOffset = id; // Update last offset
        }
        
        rs.close();
        
    } catch (SQLException e) {
        throw new ConnectException("Failed to query database", e);
    }
    
    // If no new data, return null (or empty list)
    if (records.isEmpty()) {
        Thread.sleep(pollIntervalMs); // Backoff before next poll
        return null;
    }
    
    return records;
}

Key Steps

1. Query External System: Execute SQL query, read file, call REST API, etc.
2. Iterate Results: Loop over results (rows, lines, JSON objects).
3. Create SourceRecord: For each result:
   • Source Partition: Logical identifier (e.g., {"table": "users"}).
   • Source Offset: Position in external system (e.g., {"id": 12345}).
   • Key: Kafka record key (optional, e.g., user ID).
   • Value: Kafka record value (e.g., user data).
4. Update Last Offset: Track the latest offset processed (for next poll).
5. Return Records: Return list of SourceRecord objects (or null if no data).

SourceRecord Anatomy

Constructor

public SourceRecord(
    Map<String, ?> sourcePartition,   // Source partition (e.g., {"table": "users"})
    Map<String, ?> sourceOffset,      // Source offset (e.g., {"id": 12345})
    String topic,                     // Kafka topic name
    Schema keySchema,                 // Key schema (nullable)
    Object key,                       // Key (nullable)
    Schema valueSchema,               // Value schema
    Object value                      // Value
)

Source Partition vs. Source Offset

Analogy: Source partition is like a Kafka partition, and source offset is like a Kafka offset.

Example: Multiple Tables

If a connector monitors multiple tables, each table is a source partition:

Offset Storage and Management

Reading Offsets

Use context.offsetStorageReader() to load last committed offset:

Map<String, String> sourcePartition = Collections.singletonMap("table", "users");

Map<String, Object> offset = context.offsetStorageReader()
    .offset(sourcePartition);

if (offset != null) {
    Long lastId = (Long) offset.get("id");
    // Resume from lastId
} else {
    // No previous offset, start from beginning
}

Writing Offsets (Automatic)

Connect automatically commits offsets after records are successfully produced to Kafka. You don't need to manually commit offsets in poll().

Offset Format (stored in connect-offsets topic):

Key: {"table": "users"}
Value: {"id": 12345}

Offset Commit Frequency

Configuration: offset.flush.interval.ms (worker config)

Default: 60000 (60 seconds)

Behavior: Connect commits offsets every 60 seconds (or immediately after task restart).

Schema and Struct Handling in Source Connectors

When implementing a source connector, you'll often need to convert external data (e.g., database rows, JSON objects) into Kafka Connect's internal format: Schema + Struct.

Why Schema and Struct?

Kafka Connect uses org.apache.kafka.connect.data.Schema and org.apache.kafka.connect.data.Struct to represent structured data with type information.

Benefits:

1. Type Safety: Schema defines field types (e.g., INT32, STRING, TIMESTAMP).
2. Converter Compatibility: Converters (Avro, JSON, Protobuf) use Schema to serialize data correctly.
3. Schema Evolution: Schema Registry can track schema versions and enforce compatibility.

Creating a Schema

import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;

Schema userSchema = SchemaBuilder.struct()
    .name("com.example.User")
    .version(1)
    .field("id", Schema.INT64_SCHEMA)
    .field("name", Schema.STRING_SCHEMA)
    .field("email", Schema.OPTIONAL_STRING_SCHEMA) // Optional field
    .field("created_at", org.apache.kafka.connect.data.Timestamp.SCHEMA)
    .build();

Key Methods:

• SchemaBuilder.struct(): Start building a struct schema.
• .name(): Set schema name (used by Schema Registry).
• .version(): Set schema version.
• .field(fieldName, fieldSchema): Add a field.
• Schema.OPTIONAL_*: Mark field as optional (nullable).

Common Field Types:

• Schema.INT32_SCHEMA / Schema.INT64_SCHEMA
• Schema.STRING_SCHEMA
• Schema.BOOLEAN_SCHEMA
• Schema.FLOAT64_SCHEMA
• org.apache.kafka.connect.data.Timestamp.SCHEMA
• org.apache.kafka.connect.data.Date.SCHEMA

Creating a Struct

import org.apache.kafka.connect.data.Struct;

Struct userStruct = new Struct(userSchema)
    .put("id", 12345L)
    .put("name", "Alice")
    .put("email", "alice@example.com")
    .put("created_at", new java.util.Date());

Key Methods:

• new Struct(schema): Create a struct from a schema.
• .put(fieldName, value): Set field value (must match schema type).

Example: Database Row to SourceRecord

@Override
public List<SourceRecord> poll() throws InterruptedException {
    // Query database
    ResultSet rs = statement.executeQuery("SELECT * FROM users WHERE id > " + lastId);
    
    List<SourceRecord> records = new ArrayList<>();
    
    while (rs.next()) {
        long id = rs.getLong("id");
        String name = rs.getString("name");
        String email = rs.getString("email");
        Timestamp createdAt = rs.getTimestamp("created_at");
        
        // Create Struct from row
        Struct value = new Struct(userSchema)
            .put("id", id)
            .put("name", name)
            .put("email", email)
            .put("created_at", createdAt);
        
        // Create key (simple Long)
        Long key = id;
        
        // Create SourceRecord
        SourceRecord record = new SourceRecord(
            Collections.singletonMap("table", "users"),  // source partition
            Collections.singletonMap("id", id),          // source offset
            "user-events",                               // Kafka topic
            Schema.INT64_SCHEMA,                         // key schema
            key,                                          // key
            userSchema,                                   // value schema
            value                                         // value
        );
        
        records.add(record);
        lastId = id;
    }
    
    return records;
}

Handling Null Values

Scenario: A database column allows NULL, but your schema field is non-optional.

Solution 1: Use optional schema:

.field("email", Schema.OPTIONAL_STRING_SCHEMA)

Solution 2: Provide a default value:

String email = rs.getString("email");
if (email == null) {
    email = "unknown@example.com";
}

Schema Caching

Problem: Creating a new schema on every poll() is inefficient.

Solution: Cache schema as an instance variable:

private Schema userSchema;

@Override
public void start(Map<String, String> props) {
    // Initialize schema once
    this.userSchema = SchemaBuilder.struct()
        .name("com.example.User")
        .version(1)
        .field("id", Schema.INT64_SCHEMA)
        .field("name", Schema.STRING_SCHEMA)
        .field("email", Schema.OPTIONAL_STRING_SCHEMA)
        .build();
}

@Override
public List<SourceRecord> poll() {
    // Reuse cached schema
    Struct value = new Struct(userSchema);
    // ...
}

Best Practice: Always cache schemas to avoid unnecessary object allocations.

Schema Evolution Best Practices

1. Use Schema Registry: Store schemas centrally for version management.
2. Increment Version: When adding fields, increment schema version.
3. Backward Compatibility: Add new fields as optional to maintain backward compatibility.
4. Set Schema Name: Use fully qualified names (e.g., com.example.User).

Example: Adding a Field:

// Version 1
Schema v1 = SchemaBuilder.struct()
    .name("com.example.User")
    .version(1)
    .field("id", Schema.INT64_SCHEMA)
    .field("name", Schema.STRING_SCHEMA)
    .build();

// Version 2 (adds optional field)
Schema v2 = SchemaBuilder.struct()
    .name("com.example.User")
    .version(2)
    .field("id", Schema.INT64_SCHEMA)
    .field("name", Schema.STRING_SCHEMA)
    .field("email", Schema.OPTIONAL_STRING_SCHEMA) // New field
    .build();

Poll Model and Loop Structure

Polling Frequency

How often does Connect call poll()?

• Connect calls poll() in a tight loop—as soon as poll() returns, Connect processes the records and calls poll() again.
• If poll() returns null (no data), the task should sleep before returning to avoid busy-waiting.

Example: Backoff Strategy

@Override
public List<SourceRecord> poll() throws InterruptedException {
    List<SourceRecord> records = queryExternalSystem();
    
    if (records.isEmpty()) {
        Thread.sleep(pollIntervalMs); // e.g., 5000ms
        return null;
    }
    
    return records;
}

Best Practice: Sleep for 1-10 seconds if no data is available to avoid hammering the external system.

External System Integration Patterns

Pattern 1: Incremental Pull (Database CDC Lite)

Use Case: Track new/updated rows in a database table.

Strategy: Use an incrementing column (e.g., id, updated_at) to identify new rows.

Example:

SELECT * FROM users WHERE id > <last_offset> ORDER BY id LIMIT 1000

Pros: Simple, efficient (if column is indexed).
Cons: Doesn't capture deletes, requires indexed column.

Pattern 2: Timestamp-Based Pull

Use Case: Track rows updated after a certain timestamp.

Strategy: Use a timestamp column (e.g., updated_at, modified_time).

Example:

SELECT * FROM users WHERE updated_at > <last_timestamp> ORDER BY updated_at LIMIT 1000

Pros: Captures updates, works for slowly-changing dimensions.
Cons: Doesn't capture deletes, clock skew issues.

Pattern 3: Bulk Snapshot

Use Case: One-time full table ingestion.

Strategy: Read entire table in batches.

Example:

SELECT * FROM users ORDER BY id LIMIT 1000 OFFSET <batch_offset>

Pros: Simple for initial loads.
Cons: Inefficient for large tables, doesn't track changes.

Pattern 4: File Tailing

Use Case: Stream log files as they grow (e.g., syslog, application logs).

Strategy: Track file position (byte offset) and read new lines.

Example:

File file = new File("/var/log/app.log");
RandomAccessFile raf = new RandomAccessFile(file, "r");
raf.seek(lastOffset); // Resume from last position

String line;
while ((line = raf.readLine()) != null) {
    // Create SourceRecord
    lastOffset = raf.getFilePointer(); // Update offset
}

Pros: Real-time log ingestion.
Cons: Doesn't handle log rotation (requires separate logic).

Pattern 5: REST API Polling

Use Case: Pull data from a REST API endpoint.

Strategy: Poll API with pagination and store last page offset.

Example:

String url = String.format("%s/api/users?page=%d", apiUrl, currentPage);
HttpResponse response = httpClient.get(url);
List<User> users = parseUsers(response);

for (User user : users) {
    // Create SourceRecord
}

currentPage++; // Update offset

Pros: Works for any REST API.
Cons: Requires API pagination support, slower than database queries.

Example: File Source Connector

Scenario

Read lines from a text file and produce each line as a Kafka record.

Implementation

public class FileSourceTask extends SourceTask {
    private RandomAccessFile file;
    private String filePath;
    private long fileOffset;
    private String topicName;

    @Override
    public void start(Map<String, String> props) {
        this.filePath = props.get("file.path");
        this.topicName = props.get("topic");
        
        // Load last offset
        Map<String, String> sourcePartition = Collections.singletonMap("file", filePath);
        Map<String, Object> offset = context.offsetStorageReader().offset(sourcePartition);
        
        if (offset != null) {
            this.fileOffset = (Long) offset.get("position");
        } else {
            this.fileOffset = 0L;
        }
        
        // Open file
        try {
            this.file = new RandomAccessFile(filePath, "r");
            file.seek(fileOffset); // Resume from last position
        } catch (IOException e) {
            throw new ConnectException("Failed to open file", e);
        }
    }

    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        List<SourceRecord> records = new ArrayList<>();
        
        try {
            String line;
            while ((line = file.readLine()) != null && records.size() < 1000) {
                Map<String, String> sourcePartition = Collections.singletonMap("file", filePath);
                Map<String, Long> sourceOffset = Collections.singletonMap("position", file.getFilePointer());
                
                SourceRecord record = new SourceRecord(
                    sourcePartition,
                    sourceOffset,
                    topicName,
                    Schema.STRING_SCHEMA,
                    null, // No key
                    Schema.STRING_SCHEMA,
                    line  // Line content as value
                );
                
                records.add(record);
            }
        } catch (IOException e) {
            throw new ConnectException("Failed to read file", e);
        }
        
        if (records.isEmpty()) {
            Thread.sleep(1000); // Wait 1 second if no new data
            return null;
        }
        
        return records;
    }

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

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

Key Takeaways

1. SourceTask.poll() is the core method—query external system, convert to SourceRecord, return list.
2. Source Partition identifies logical partition (e.g., table, file); Source Offset tracks position.
3. Offset Management: Use context.offsetStorageReader() to load offsets; Connect commits automatically.
4. Poll Loop: Connect calls poll() repeatedly; sleep if no data to avoid busy-waiting.
5. External Integration: Incremental pull, timestamp-based, bulk snapshot, file tailing, REST API polling.

What's Next?

• Module 7.3 covers Sink Connector implementation.
• Module 7.4 explains testing and packaging custom connectors.
• Module 2.1 reviews Source vs. Sink semantics (cross-reference).

Further Reading

• SourceTask API Docs
• Source Connector Tutorial (Confluent)

End of Module 7.2