7.4 – Testing & Packaging Connectors

Overview

This module covers how to test and package custom Kafka Connect connectors for production deployment. By the end, you will understand:

Unit testing connectors and tasks.
Integration testing with embedded Kafka and external systems.
Packaging connectors as JARs with dependencies.
Deployment practices for production.

This module completes the custom connector development workflow. Modules 7.1-7.3 covered implementation; this module focuses on quality assurance and deployment.

Testing Kafka Connect Connectors

Testing Pyramid for Connectors

       ┌────────────────┐
       │ E2E Tests      │  (Few tests, slow, full cluster)
       ├────────────────┤
       │ Integration    │  (Some tests, medium speed, embedded Kafka)
       ├────────────────┤
       │ Unit Tests     │  (Many tests, fast, mocked dependencies)
       └────────────────┘

Best Practice: Prioritize unit tests, add integration tests for critical paths, use E2E tests sparingly.

Unit Testing

What to Test

Connector Configuration: Validate ConfigDef and config parsing.
Task Configuration: Verify taskConfigs() splits work correctly.
Task Logic: Test poll() (source) or put() (sink) with mocked external systems.

Example: Test Connector Configuration

@Test
public void testValidConfiguration() {
    Map<String, String> config = new HashMap<>();
    config.put("connection.url", "jdbc:postgresql://localhost:5432/mydb");
    config.put("table.whitelist", "users,orders");
    config.put("poll.interval.ms", "5000");
    
    MyConnector connector = new MyConnector();
    connector.start(config);
    
    assertNotNull(connector.taskClass());
}

@Test(expected = ConnectException.class)
public void testMissingRequiredConfig() {
    Map<String, String> config = new HashMap<>();
    // Missing connection.url
    
    MyConnector connector = new MyConnector();
    connector.start(config); // Should throw ConnectException
}

Example: Test Task Configuration Splitting

@Test
public void testTaskConfigsSplitsWorkCorrectly() {
    Map<String, String> config = new HashMap<>();
    config.put("table.whitelist", "users,orders,products");
    
    MyConnector connector = new MyConnector();
    connector.start(config);
    
    List<Map<String, String>> taskConfigs = connector.taskConfigs(3);
    
    assertEquals(3, taskConfigs.size());
    assertEquals("users", taskConfigs.get(0).get("table.name"));
    assertEquals("orders", taskConfigs.get(1).get("table.name"));
    assertEquals("products", taskConfigs.get(2).get("table.name"));
}

Example: Test Source Task Poll Logic (Mocked)

@Test
public void testSourceTaskPoll() {
    // Mock external system
    ExternalSystem mockSystem = mock(ExternalSystem.class);
    when(mockSystem.fetchData()).thenReturn(Arrays.asList(
        new Data(1, "Alice"),
        new Data(2, "Bob")
    ));
    
    // Create and start task
    MySourceTask task = new MySourceTask();
    task.setExternalSystem(mockSystem); // Inject mock
    
    Map<String, String> taskConfig = new HashMap<>();
    taskConfig.put("topic", "test-topic");
    task.start(taskConfig);
    
    // Poll for records
    List<SourceRecord> records = task.poll();
    
    assertNotNull(records);
    assertEquals(2, records.size());
    assertEquals("test-topic", records.get(0).topic());
    assertEquals("Alice", ((Struct) records.get(0).value()).getString("name"));
}

Example: Test Sink Task Put Logic (Mocked)

@Test
public void testSinkTaskPut() {
    // Mock external system
    ExternalSystem mockSystem = mock(ExternalSystem.class);
    
    // Create and start task
    MySinkTask task = new MySinkTask();
    task.setExternalSystem(mockSystem); // Inject mock
    
    Map<String, String> taskConfig = new HashMap<>();
    task.start(taskConfig);
    
    // Create test records
    Struct value = new Struct(schema)
        .put("id", 1L)
        .put("name", "Alice");
    
    SinkRecord record = new SinkRecord(
        "test-topic", 0, Schema.STRING_SCHEMA, "1", schema, value, 0L
    );
    
    // Call put()
    task.put(Collections.singletonList(record));
    
    // Verify external system was called
    verify(mockSystem, times(1)).write(any());
}

Integration Testing

What to Test

End-to-End Flow: Connector reads from external system, writes to Kafka (or vice versa).
Offset Management: Verify offsets are committed and restored correctly.
Error Handling: Test retry behavior, DLQ, poison records.

Tools

Embedded Kafka: kafka-clients test utilities or Testcontainers.
Testcontainers: Docker-based integration testing (PostgreSQL, Elasticsearch, etc.).

Example: Embedded Kafka + Testcontainers

Dependencies (Maven)

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.5.0</version>
    <classifier>test</classifier>
    <scope>test</scope>
</dependency>

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>testcontainers</artifactId>
    <version>1.19.0</version>
    <scope>test</scope>
</dependency>

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>kafka</artifactId>
    <version>1.19.0</version>
    <scope>test</scope>
</dependency>

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>postgresql</artifactId>
    <version>1.19.0</version>
    <scope>test</scope>
</dependency>

Integration Test Example

@Test
public void testJdbcSourceToKafka() {
    // Start PostgreSQL container
    PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:15")
        .withDatabaseName("testdb")
        .withUsername("test")
        .withPassword("test");
    postgres.start();
    
    // Start Kafka container
    KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.5.0"));
    kafka.start();
    
    // Insert test data into PostgreSQL
    try (Connection conn = DriverManager.getConnection(postgres.getJdbcUrl(), "test", "test")) {
        Statement stmt = conn.createStatement();
        stmt.execute("CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR)");
        stmt.execute("INSERT INTO users VALUES (1, 'Alice'), (2, 'Bob')");
    }
    
    // Configure and start connector
    Map<String, String> connectorConfig = new HashMap<>();
    connectorConfig.put("connection.url", postgres.getJdbcUrl());
    connectorConfig.put("table.whitelist", "users");
    connectorConfig.put("topic.prefix", "postgres-");
    connectorConfig.put("mode", "bulk");
    
    MyConnector connector = new MyConnector();
    connector.start(connectorConfig);
    
    MySourceTask task = new MySourceTask();
    task.start(connectorConfig);
    
    // Poll records
    List<SourceRecord> records = task.poll();
    
    assertNotNull(records);
    assertEquals(2, records.size());
    assertEquals("postgres-users", records.get(0).topic());
    
    // Cleanup
    task.stop();
    connector.stop();
    postgres.stop();
    kafka.stop();
}

Packaging Connectors

Project Structure

my-kafka-connector/
├── pom.xml (or build.gradle)
├── src/
│   ├── main/
│   │   ├── java/
│   │   │   └── com/example/
│   │   │       ├── MyConnector.java
│   │   │       ├── MySourceTask.java
│   │   │       └── MyConnectorConfig.java
│   │   └── resources/
│   │       └── manifest.json (optional, for Confluent Hub)
│   └── test/
│       └── java/
│           └── com/example/
│               ├── MyConnectorTest.java
│               └── MySourceTaskTest.java
└── README.md

Maven Configuration

`pom.xml`

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
                             http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>my-kafka-connector</artifactId>
    <version>1.0.0</version>
    <packaging>jar</packaging>

    <dependencies>
        <!-- Kafka Connect API -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>connect-api</artifactId>
            <version>3.5.0</version>
            <scope>provided</scope>
        </dependency>
        
        <!-- External system client (e.g., JDBC driver) -->
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
            <version>42.5.0</version>
        </dependency>
        
        <!-- Testing -->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.2</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!-- Maven Shade Plugin: Create fat JAR with dependencies -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <manifestEntries>
                                        <Connect-Version>3.5.0</Connect-Version>
                                    </manifestEntries>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

Build and Package

mvn clean package

Output: target/my-kafka-connector-1.0.0.jar (fat JAR with dependencies)

Deployment Practices

1. Deploy to Plugin Path

Worker Configuration (connect-distributed.properties):

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

Deployment:

mkdir -p /usr/local/share/kafka/plugins/my-kafka-connector
cp target/my-kafka-connector-1.0.0.jar /usr/local/share/kafka/plugins/my-kafka-connector/

Restart Workers:

bin/kafka-server-stop.sh
bin/connect-distributed.sh config/connect-distributed.properties

2. Verify Plugin Installation

Check Available Plugins:

curl http://localhost:8083/connector-plugins | jq

Expected Output:

[
  {
    "class": "com.example.MyConnector",
    "type": "source",
    "version": "1.0.0"
  },
  ...
]

2a. Installing External Plugins from Confluent Hub

For production use, you'll often need to install pre-built connectors from Confluent Hub. Here's how:

Method 1: Using Confluent Hub Client

Install Confluent Hub Client:

# Download and install Confluent Platform (includes confluent-hub CLI)
curl -O http://packages.confluent.io/archive/7.5/confluent-community-7.5.0.tar.gz
tar xzf confluent-community-7.5.0.tar.gz
export PATH=$PATH:$(pwd)/confluent-7.5.0/bin

Install a Connector:

confluent-hub install confluentinc/kafka-connect-jdbc:10.7.4

What This Does:

Downloads the connector plugin from Confluent Hub.
Extracts to /usr/share/confluent-hub-components/confluentinc-kafka-connect-jdbc/.
Automatically sets up plugin path.

Verify Installation:

ls /usr/share/confluent-hub-components/
# Output: confluentinc-kafka-connect-jdbc/

Restart Workers:

bin/connect-distributed.sh config/connect-distributed.properties

Method 2: Manual Download and Installation

Step 1: Download from Confluent Hub

Visit Confluent Hub and download the ZIP file for your connector (e.g., confluentinc-kafka-connect-jdbc-10.7.4.zip).

Step 2: Extract to Plugin Path

unzip confluentinc-kafka-connect-jdbc-10.7.4.zip -d /usr/local/share/kafka/plugins/

Step 3: Verify Directory Structure

ls /usr/local/share/kafka/plugins/confluentinc-kafka-connect-jdbc/
# Expected output: lib/, doc/, manifest.json, etc/

Step 4: Configure Worker Plugin Path

Edit connect-distributed.properties:

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

Step 5: Restart Workers and Verify

bin/connect-distributed.sh config/connect-distributed.properties &
sleep 10
curl http://localhost:8083/connector-plugins | jq '.[] | select(.class | contains("JdbcSource"))'

Expected Output:

{
  "class": "io.confluent.connect.jdbc.JdbcSourceConnector",
  "type": "source",
  "version": "10.7.4"
}

Method 3: Docker-Based Plugin Installation

For containerized deployments, install plugins at build time or runtime.

Build-Time Installation (Dockerfile):

FROM confluentinc/cp-kafka-connect:7.5.0

# Install plugin using confluent-hub
RUN confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:10.7.4
RUN confluent-hub install --no-prompt debezium/debezium-connector-mysql:2.4.0

# Or manually copy plugin
COPY my-custom-connector/ /usr/share/confluent-hub-components/my-custom-connector/

Build and Run:

docker build -t my-kafka-connect:latest .
docker run -d \
  --name kafka-connect \
  -p 8083:8083 \
  -e CONNECT_BOOTSTRAP_SERVERS=kafka:9092 \
  -e CONNECT_GROUP_ID=connect-cluster \
  -e CONNECT_CONFIG_STORAGE_TOPIC=connect-configs \
  -e CONNECT_OFFSET_STORAGE_TOPIC=connect-offsets \
  -e CONNECT_STATUS_STORAGE_TOPIC=connect-status \
  -e CONNECT_KEY_CONVERTER=org.apache.kafka.connect.json.JsonConverter \
  -e CONNECT_VALUE_CONVERTER=org.apache.kafka.connect.json.JsonConverter \
  -e CONNECT_PLUGIN_PATH=/usr/share/java,/usr/share/confluent-hub-components \
  my-kafka-connect:latest

Runtime Installation (Mount Volume):

# Download and extract plugin on host
confluent-hub install --component-dir ./plugins confluentinc/kafka-connect-jdbc:10.7.4

# Mount plugin directory into container
docker run -d \
  --name kafka-connect \
  -p 8083:8083 \
  -v $(pwd)/plugins:/usr/share/confluent-hub-components \
  -e CONNECT_PLUGIN_PATH=/usr/share/confluent-hub-components \
  confluentinc/cp-kafka-connect:7.5.0

Method 4: Kubernetes-Based Plugin Installation

For Kubernetes deployments, use init containers to download plugins.

Example: Kubernetes StatefulSet

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka-connect
spec:
  serviceName: kafka-connect
  replicas: 3
  template:
    spec:
      initContainers:
      - name: plugin-installer
        image: confluentinc/cp-kafka-connect:7.5.0
        command:
        - sh
        - -c
        - |
          confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:10.7.4
          confluent-hub install --no-prompt debezium/debezium-connector-mysql:2.4.0
        volumeMounts:
        - name: plugins
          mountPath: /usr/share/confluent-hub-components
      containers:
      - name: kafka-connect
        image: confluentinc/cp-kafka-connect:7.5.0
        ports:
        - containerPort: 8083
        env:
        - name: CONNECT_PLUGIN_PATH
          value: "/usr/share/java,/usr/share/confluent-hub-components"
        volumeMounts:
        - name: plugins
          mountPath: /usr/share/confluent-hub-components
      volumes:
      - name: plugins
        emptyDir: {}

What This Does:

Init Container runs before main container, installs plugins.
Shared Volume (emptyDir) makes plugins available to main container.
Main Container starts with plugins pre-installed.

3. Deploy via Confluent Hub (Optional)

Create manifest.json:

{
  "name": "my-kafka-connector",
  "version": "1.0.0",
  "title": "My Custom Kafka Connector",
  "description": "Connector for integrating with MySystem",
  "owner": {
    "name": "Your Name",
    "url": "https://example.com"
  },
  "tags": ["database", "source"],
  "requirements": ["Kafka 3.5+"],
  "documentation_url": "https://example.com/docs",
  "component_types": ["source"]
}

Install via Confluent Hub CLI:

confluent-hub install target/my-kafka-connector-1.0.0.zip

Production Readiness Checklist

Configuration

All required configs have defaults or validation errors.
Sensitive configs (passwords) support externalized secrets.
ConfigDef includes documentation for all properties.

Testing

Unit tests cover connector and task logic (>80% coverage).
Integration tests verify end-to-end flow with real systems.
Error scenarios tested (network failures, invalid data, etc.).

Performance

Connector supports task parallelism (tasks.max > 1).
Batch writes implemented for sink connectors.
No memory leaks or resource leaks (verified via profiling).

Error Handling

Transient errors trigger retries (configurable timeout).
Permanent errors logged and/or sent to DLQ.
Poison records don't block the pipeline.

Monitoring

JMX metrics exposed for throughput, errors, latency.
Logs include actionable error messages.
REST API status endpoint returns meaningful state.

Documentation

README includes setup instructions, config examples, troubleshooting.
Javadoc for all public classes and methods.
Changelog documents version history.

Key Takeaways

Unit tests cover config validation, task splitting, and core logic (with mocked dependencies).
Integration tests verify end-to-end flow with embedded Kafka and Testcontainers.
Packaging: Use Maven Shade Plugin to create fat JARs with dependencies.
Deployment: Copy JAR to plugin.path, restart workers, verify via REST API.
Production readiness: Test error scenarios, monitor metrics, document thoroughly.

What's Next?

Module 8.1 covers monitoring and metrics for connectors in production.
Module 9.1 explains security best practices for Connect deployments.
Module 6.1 reviews error handling (critical for production connectors).

7.4 – Testing & Packaging Connectors

Overview

Testing Kafka Connect Connectors

Testing Pyramid for Connectors

Unit Testing

What to Test

Example: Test Connector Configuration

Example: Test Task Configuration Splitting

Example: Test Source Task Poll Logic (Mocked)

Example: Test Sink Task Put Logic (Mocked)

Integration Testing

What to Test

Tools

Example: Embedded Kafka + Testcontainers

Dependencies (Maven)

Integration Test Example

Packaging Connectors

Project Structure

Maven Configuration

pom.xml

Build and Package

Deployment Practices

1. Deploy to Plugin Path

2. Verify Plugin Installation

2a. Installing External Plugins from Confluent Hub

Method 1: Using Confluent Hub Client

Method 2: Manual Download and Installation

Method 3: Docker-Based Plugin Installation

Method 4: Kubernetes-Based Plugin Installation

3. Deploy via Confluent Hub (Optional)

Production Readiness Checklist

Configuration

Testing

Performance

Error Handling

Monitoring

Documentation

Key Takeaways

What's Next?

Further Reading

`pom.xml`