KafkaIQ

7.1 – Connector Development Model

Overview

This module introduces the Kafka Connect plugin model for developing custom connectors. By the end, you will understand:

  • The Connector API and Task API.
  • Connector and Task lifecycle methods.
  • Configuration definitions (ConfigDef).
  • Plugin isolation and classloading.

This module establishes the foundation for custom connector development. Modules 7.2-7.4 will dive into Source and Sink implementation details.

Why Develop a Custom Connector?

When to Build a Custom Connector

✅ Build a Custom Connector When:

  • No existing connector supports your source/sink system.
  • Existing connectors lack critical features for your use case.
  • You need fine-grained control over data movement logic.

❌ Avoid Building a Custom Connector When:

  • An existing connector already exists (check Confluent Hub).
  • The integration can be achieved with a custom producer/consumer + SMTs.

Cost-Benefit: Developing a connector requires upfront investment (implementation, testing, maintenance), but yields long-term benefits (reusability, operational simplicity).

Kafka Connect Plugin Architecture

Key Components

  1. Connector: Defines the job, manages configuration, and creates tasks.
  2. Task: Executes the actual data movement (read from source, write to sink).
  3. ConfigDef: Declares configuration properties and validation rules.

Separation of Concerns

Connector (Management Layer)
    ├── Validates configuration
    ├── Splits work into tasks
    └── Monitors task health

Task (Execution Layer)
    ├── Reads/writes data
    ├── Commits offsets
    └── Reports progress

Key Insight: Connectors are stateless—they don't process data directly. Tasks are stateful and do the actual work.

The Connector API

Interface: org.apache.kafka.connect.connector.Connector

All connectors extend either:

  • SourceConnector: For source connectors (external system → Kafka).
  • SinkConnector: For sink connectors (Kafka → external system).

Key Methods

1. start(Map<String, String> props)

Purpose: Initialize the connector with configuration properties.

Called: Once, when the connector is created.

Responsibilities:

  • Validate configuration.
  • Initialize connection to external system (if needed).
  • Store configuration for later use.

Example:

@Override
public void start(Map<String, String> props) {
    this.config = new MyConnectorConfig(props);
    // Validate connection to external system
    try {
        testConnection(config);
    } catch (Exception e) {
        throw new ConnectException("Failed to connect to external system", e);
    }
}

2. Class<? extends Task> taskClass()

Purpose: Return the task class that will execute data movement.

Called: When the connector needs to create tasks.

Example:

@Override
public Class<? extends Task> taskClass() {
    return MySourceTask.class;
}

3. List<Map<String, String>> taskConfigs(int maxTasks)

Purpose: Split work into multiple tasks by generating task-specific configurations.

Called: When the connector is started or reconfigured.

Parameters:

  • maxTasks: Maximum number of tasks (from tasks.max config).

Returns: List of configuration maps (one per task).

Example (JDBC Source: one task per table):

@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
    List<String> tables = config.getTables(); // e.g., ["users", "orders", "products"]
    List<Map<String, String>> taskConfigs = new ArrayList<>();
    
    for (int i = 0; i < Math.min(maxTasks, tables.size()); i++) {
        Map<String, String> taskConfig = new HashMap<>(config.originalsStrings());
        taskConfig.put("table.name", tables.get(i));
        taskConfigs.add(taskConfig);
    }
    
    return taskConfigs;
}

Result: If maxTasks=3 and tables=["users", "orders", "products"], create 3 tasks:

  • Task 0: {"table.name": "users"}
  • Task 1: {"table.name": "orders"}
  • Task 2: {"table.name": "products"}

4. void stop()

Purpose: Clean up resources when the connector is stopped.

Called: When the connector is deleted or the worker shuts down.

Example:

@Override
public void stop() {
    // Close database connections, HTTP clients, etc.
    if (connection != null) {
        connection.close();
    }
}

5. ConfigDef config()

Purpose: Define connector configuration properties (name, type, default, validation).

Called: When validating connector configuration via REST API.

Example:

@Override
public ConfigDef config() {
    return new ConfigDef()
        .define("connection.url", ConfigDef.Type.STRING, ConfigDef.NO_DEFAULT_VALUE,
                ConfigDef.Importance.HIGH, "Database connection URL")
        .define("table.whitelist", ConfigDef.Type.LIST, Collections.emptyList(),
                ConfigDef.Importance.MEDIUM, "Tables to monitor")
        .define("poll.interval.ms", ConfigDef.Type.INT, 5000,
                ConfigDef.Range.atLeast(100), ConfigDef.Importance.LOW,
                "Polling interval in milliseconds");
}

6. String version()

Purpose: Return the connector version (for logging and monitoring).

Example:

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

The Task API

Interface: org.apache.kafka.connect.connector.Task

All tasks extend either:

  • SourceTask: For source connectors (poll external system, produce to Kafka).
  • SinkTask: For sink connectors (consume from Kafka, write to external system).

Key Methods (Common to Source and Sink)

1. void start(Map<String, String> props)

Purpose: Initialize the task with task-specific configuration.

Called: Once, when the task is started.

Responsibilities:

  • Parse task configuration.
  • Connect to external system.
  • Initialize state (e.g., last offset read).

Example:

@Override
public void start(Map<String, String> props) {
    this.config = new MyTaskConfig(props);
    this.connection = connectToExternalSystem(config);
    this.lastOffset = loadLastOffset();
}

2. void stop()

Purpose: Clean up resources when the task is stopped.

Called: When the task is reassigned, the connector is deleted, or the worker shuts down.

Example:

@Override
public void stop() {
    if (connection != null) {
        connection.close();
    }
}

3. String version()

Purpose: Return the task version (must match connector version).

Example:

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

Configuration Definitions (ConfigDef)

Purpose

ConfigDef defines:

  • Configuration property names.
  • Data types (STRING, INT, BOOLEAN, LIST, etc.).
  • Default values.
  • Validation rules (e.g., range, regex, custom validators).
  • Documentation (displayed in REST API validation).

Example: Full ConfigDef

public static ConfigDef configDef() {
    return new ConfigDef()
        // Required property
        .define("connection.url", ConfigDef.Type.STRING, ConfigDef.NO_DEFAULT_VALUE,
                ConfigDef.Importance.HIGH, "Database connection URL")
        
        // Optional property with default
        .define("poll.interval.ms", ConfigDef.Type.INT, 5000,
                ConfigDef.Range.atLeast(100), ConfigDef.Importance.MEDIUM,
                "Polling interval in milliseconds")
        
        // List property
        .define("table.whitelist", ConfigDef.Type.LIST, Collections.emptyList(),
                ConfigDef.Importance.HIGH, "Tables to monitor (comma-separated)")
        
        // Boolean property
        .define("auto.create", ConfigDef.Type.BOOLEAN, false,
                ConfigDef.Importance.LOW, "Auto-create missing topics")
        
        // Custom validator
        .define("batch.size", ConfigDef.Type.INT, 1000,
                new ConfigDef.Range(1, 10000), ConfigDef.Importance.MEDIUM,
                "Batch size (1-10000)");
}

Property Importance Levels

Importance Description
HIGH Required or critical properties (e.g., connection URL)
MEDIUM Important but optional properties (e.g., batch size)
LOW Advanced or rarely changed properties (e.g., timeouts)

Plugin Isolation and Classloading

What is Plugin Isolation?

Plugin isolation ensures that connector JARs (and their dependencies) are loaded in isolated classloaders, preventing dependency conflicts.

Example: If Connector A uses Jackson 2.10 and Connector B uses Jackson 2.12, both can coexist without conflicts.

Plugin Path

Configuration: plugin.path in worker config.

Purpose: Directory where connector JARs are located.

Example:

plugin.path=/usr/local/share/kafka/plugins

Directory Structure:

/usr/local/share/kafka/plugins/
├── jdbc-connector/
│   ├── kafka-connect-jdbc-10.5.0.jar
│   └── postgresql-42.3.1.jar
└── elasticsearch-connector/
    ├── kafka-connect-elasticsearch-11.1.0.jar
    └── elasticsearch-rest-client-7.15.0.jar

Key Rule: Each connector should be in its own subdirectory to enable plugin isolation.

Connector Lifecycle

1. Deployment

Operator submits connector configuration via REST API:

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{"name": "my-connector", "config": {...}}'

2. Connector Initialization

  1. Connect worker validates configuration using ConfigDef.
  2. Worker calls Connector.start(props) to initialize the connector.
  3. Connector validates connection to external system.

3. Task Creation

  1. Worker calls Connector.taskConfigs(maxTasks) to generate task configurations.
  2. Worker assigns tasks to workers (in distributed mode).

4. Task Execution

  1. Worker calls Task.start(props) for each task.
  2. Tasks begin processing data (source: poll(), sink: put()).

5. Rebalancing

If workers join/leave or tasks.max changes:

  1. Worker calls Task.stop() for reassigned tasks.
  2. Worker calls Connector.taskConfigs(maxTasks) to recompute task assignments.
  3. Worker calls Task.start(props) for newly assigned tasks.

6. Connector Deletion

  1. Worker calls Task.stop() for all tasks.
  2. Worker calls Connector.stop() to clean up resources.

Example: Minimal Source Connector

Connector Class

public class MySourceConnector extends SourceConnector {
    private Map<String, String> config;

    @Override
    public void start(Map<String, String> props) {
        this.config = props;
    }

    @Override
    public Class<? extends Task> taskClass() {
        return MySourceTask.class;
    }

    @Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        List<Map<String, String>> configs = new ArrayList<>();
        for (int i = 0; i < maxTasks; i++) {
            configs.add(new HashMap<>(config));
        }
        return configs;
    }

    @Override
    public void stop() {
        // Cleanup
    }

    @Override
    public ConfigDef config() {
        return new ConfigDef()
            .define("topic", ConfigDef.Type.STRING, ConfigDef.NO_DEFAULT_VALUE,
                    ConfigDef.Importance.HIGH, "Target Kafka topic");
    }

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

Key Takeaways

  1. Connector API: Manages configuration, splits work, and creates tasks (start, taskConfigs, stop).
  2. Task API: Executes data movement (start, poll/put, stop).
  3. ConfigDef: Declares configuration properties, types, defaults, and validation.
  4. Plugin Isolation: Connectors are loaded in isolated classloaders to prevent dependency conflicts.
  5. Lifecycle: Deployment → Connector Init → Task Creation → Execution → Rebalancing → Deletion.

What's Next?

  • Module 7.2 covers Source Connector implementation in detail.
  • Module 7.3 covers Sink Connector implementation in detail.
  • Module 7.4 explains testing and packaging custom connectors.

Further Reading

End of Module 7.1