3.3 – Task Parallelism & Scaling

Overview

This module is the authoritative explanation of Kafka Connect's scaling model. By the end, you will understand:

• The relationship between connectors and tasks.
• How task parallelism works.
• Partition-to-task mapping for source and sink connectors.
• Task rebalancing behavior.
• Scaling limits and best practices.

All concepts about scaling and parallelism live here. Other modules (e.g., performance tuning) will reference this module, not re-explain these mechanics.

Connectors vs. Tasks: The Relationship

Conceptual Model

Connector (Logical Job)
    │
    ├── Task 1 (Execution Unit)
    ├── Task 2 (Execution Unit)
    ├── Task 3 (Execution Unit)
    └── Task N (Execution Unit)

Key Insight: A connector is a job definition that doesn't process data itself. Instead, it spawns tasks that do the actual work.

Connector Responsibilities

1. Define Configuration: Connection details, table names, topics, etc.
2. Split Work: Determine how to divide work into tasks.
3. Monitor Tasks: Track task health and request restarts if needed.

Task Responsibilities

1. Execute Work: Read from external systems (source) or write to external systems (sink).
2. Report Progress: Commit offsets, report status to the Connect runtime.

Analogy: A connector is like a project manager that defines a job and breaks it into subtasks. Tasks are the workers who execute those subtasks.

Task Parallelism

What is Task Parallelism?

Task parallelism is the ability to run multiple tasks concurrently to increase throughput.

Configured via tasks.max

The maximum number of tasks a connector can create is controlled by the tasks.max configuration:

{
  "name": "jdbc-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "5",
    ...
  }
}

Important: tasks.max is a maximum, not a guarantee. The actual number of tasks created depends on:

1. Connector Logic: The connector plugin determines how to split work.
2. Available Work: For example, a JDBC source with 3 tables can only create 3 tasks, even if tasks.max=10.

Example: JDBC Source with 3 Tables

Configuration:

{
  "tasks.max": "5",
  "table.whitelist": "users,orders,products"
}

Result: Only 3 tasks are created (one per table), even though tasks.max=5.

Partition-to-Task Mapping

Source Connectors

For source connectors, the connector plugin determines how to partition work. Common strategies:

1. Per-Table Partitioning (JDBC Source)

Each task reads from one or more database tables.

Example: tasks.max=2, table.whitelist=users,orders,products

2. Per-File Partitioning (File Source)

Each task reads from one or more files.

Example: tasks.max=3, monitoring directory with 5 files.

3. Range Partitioning (Custom Source)

Tasks read different ranges of data (e.g., user IDs 1-1000, 1001-2000, etc.).

Sink Connectors

For sink connectors, tasks are typically assigned based on Kafka partition assignment.

Kafka Partition Assignment

Each sink task consumes from one or more Kafka partitions.

Example: Topic orders with 6 partitions, tasks.max=3

Key Rule: The number of sink tasks cannot exceed the number of Kafka partitions. If tasks.max=10 but the topic has only 3 partitions, only 3 tasks will be created.

Task Distribution Across Workers

In Distributed Mode

Tasks are distributed across workers in a Connect cluster.

Example: 3 workers, 6 tasks

Rebalancing: If a worker fails, its tasks are reassigned to surviving workers.

In Standalone Mode

All tasks run on the single worker.

Task Rebalancing

When Does Rebalancing Happen?

Rebalancing is triggered by:

1. Worker Joins: New worker added to the cluster.
2. Worker Leaves: Worker crashes or is shut down.
3. Connector Reconfiguration: tasks.max is changed.
4. Task Failure: A task crashes and needs reassignment.

Rebalance Process

1. Detection: Workers detect membership changes via Kafka's group coordination protocol.
2. Leader Election: The group leader computes a new task assignment.
3. Task Stop: Workers stop tasks that are no longer assigned to them.
4. Task Start: Workers start newly assigned tasks.

Duration: Typically 5-10 seconds in a healthy cluster. During this time, affected tasks are paused.

Example: Worker Failure

Before Failure:

After Worker 2 Fails:

Impact: Tasks 2 and 3 are briefly paused during reassignment, then resume on Worker 1 and Worker 3.

Scaling Strategies

Horizontal Scaling (Add More Tasks)

How: Increase tasks.max in connector configuration.

When to Use:

• Connector throughput is bottlenecked by task processing speed.
• External system can handle more concurrent connections (e.g., database has available connection pool capacity).

Example: JDBC source reading from a large table.

{
  "tasks.max": "10"  // Increase from 3 to 10
}

Limitation: Cannot exceed the number of logical partitions (e.g., database tables, Kafka partitions).

Vertical Scaling (Add More Workers)

How: Add more workers to the Connect cluster.

When to Use:

• Existing workers are CPU/memory-saturated.
• You want higher fault tolerance (more workers = more redundancy).

Example: Scale from 3 workers to 5 workers.

Benefit: Tasks are redistributed across more workers, reducing per-worker load.

Combined Scaling

Best Practice: Scale both tasks and workers together for maximum throughput.

Example: Increase tasks.max from 3 to 12 and add 2 more workers.

Scaling Limits

1. Connector-Defined Limits

Some connectors cannot create more tasks than logical partitions:

• JDBC Source: Max tasks = number of tables.
• File Source: Max tasks = number of files.
• Kafka Sink: Max tasks = number of Kafka partitions.

2. External System Limits

• Database Connection Pools: If a database allows 10 concurrent connections, creating 20 tasks may cause connection errors.
• API Rate Limits: REST API-based sources may throttle requests, limiting effective parallelism.

3. Worker Capacity

Each worker has finite CPU/memory. Too many tasks on one worker can degrade performance.

Rule of Thumb: Aim for 5-10 tasks per worker, depending on task complexity.

4. Kafka Partition Count

For sink connectors, the number of tasks is capped by the number of Kafka partitions.

Example: Topic with 3 partitions → max 3 sink tasks, even if tasks.max=10.

Practical Scaling Examples

Example 1: Scaling a JDBC Source

Scenario: Ingest data from 10 database tables.

Configuration:

{
  "tasks.max": "10",
  "table.whitelist": "users,orders,products,..."
}

Result: 10 tasks (1 per table), distributed across workers.

Bottleneck: Database connection pool (ensure pool size ≥ 10).

Example 2: Scaling a Kafka Sink

Scenario: Write data to Elasticsearch from a Kafka topic with 12 partitions.

Configuration:

{
  "tasks.max": "12",
  "topics": "logs"
}

Result: 12 tasks (1 per partition), distributed across workers.

Bottleneck: Elasticsearch indexing throughput (tune batch.size and max.buffered.records).

Example 3: Over-Provisioning

Scenario: Set tasks.max=20 but only 5 tables exist.

Result: Only 5 tasks are created (connector logic limits task count).

Lesson: Setting tasks.max too high has no negative effect (but no benefit either).

Task Affinity and Stickiness

What is Task Affinity?

Task affinity is the preference to assign a task to the same worker it previously ran on.

Benefits:

• Faster Startup: If state stores or cached data exist on the worker, tasks start faster.
• Reduced Data Transfer: For stateful connectors, local state avoids network overhead.

Implementation: Connect uses a "sticky" assignment strategy where tasks prefer their previous worker during rebalancing.

When Affinity is Broken

Affinity is broken when:

1. The previous worker is no longer available (crash, decommission).
2. The cluster is rebalanced due to worker additions.

Impact: Tasks may take slightly longer to start due to state reconstruction or cache misses.

Monitoring Task Parallelism

JMX Metrics

Connector-Level Metrics:

• kafka.connect:type=connector-metrics,connector="{connector-name}" → connector-task-count

Task-Level Metrics:

• kafka.connect:type=task-metrics,connector="{connector-name}",task="{task-id}" → batch-size-avg, offset-commit-completion-rate

REST API

Check Task Assignment:

curl http://localhost:8083/connectors/jdbc-source/status

Response:

{
  "name": "jdbc-source",
  "connector": {"state": "RUNNING", "worker_id": "worker1:8083"},
  "tasks": [
    {"id": 0, "state": "RUNNING", "worker_id": "worker1:8083"},
    {"id": 1, "state": "RUNNING", "worker_id": "worker2:8083"},
    {"id": 2, "state": "RUNNING", "worker_id": "worker3:8083"}
  ]
}

Interpretation: 3 tasks running across 3 workers.

Best Practices for Scaling

1. Start Small, Scale Gradually

Begin with tasks.max=1, verify functionality, then increase incrementally.

2. Match Tasks to Logical Partitions

For JDBC sources, set tasks.max equal to the number of tables (if feasible).

3. Monitor External System Load

Ensure the external system (database, API, etc.) can handle the increased concurrency.

4. Use Multiple Workers for Fault Tolerance

Even if tasks.max=1, use ≥3 workers for high availability.

5. Tune Kafka Partition Count

For sink connectors, ensure the Kafka topic has enough partitions to support desired parallelism.

Example: If you want 10 sink tasks, the topic should have ≥10 partitions.

6. Avoid Over-Scaling

More tasks ≠ always better. Too many tasks can:

• Overwhelm external systems.
• Increase coordination overhead.
• Degrade per-task throughput.

Common Scaling Pitfalls

Pitfall 1: tasks.max Exceeds Logical Partitions

Symptom: tasks.max=10, but only 3 tasks are created.

Cause: Connector can only create tasks for available work (e.g., 3 tables, 3 files).

Fix: Align tasks.max with the actual number of partitions.

Pitfall 2: Database Connection Pool Exhaustion

Symptom: Tasks fail with "Too many connections" errors.

Cause: tasks.max exceeds database connection pool size.

Fix: Increase database connection pool or reduce tasks.max.

Pitfall 3: Kafka Partition Bottleneck

Symptom: Sink connector with tasks.max=10, but throughput doesn't increase.

Cause: Kafka topic has only 3 partitions → only 3 tasks are created.

Fix: Increase Kafka partition count (requires topic reconfiguration or new topic).

Pitfall 4: Unbalanced Task Distribution

Symptom: Some workers are idle while others are overloaded.

Cause: Connector logic creates uneven task assignments.

Fix: Ensure logical partitions (tables, files, etc.) are balanced in size.

Key Takeaways

1. Connectors define jobs; tasks execute them. Tasks are units of parallelism.
2. tasks.max controls maximum task count, but actual count depends on connector logic.
3. Source connectors partition by logical units (tables, files); sink connectors partition by Kafka partitions.
4. Rebalancing redistributes tasks when workers join/leave or connectors are reconfigured.
5. Scaling strategies: Increase tasks.max (horizontal), add workers (vertical), or both.
6. Scaling limits: Logical partitions, external system capacity, worker capacity, Kafka partitions.
7. Best practice: Start small, scale gradually, and monitor external system load.

What's Next?

• Module 4.1 covers Single Message Transforms (SMTs) for record-level transformations.
• Module 5.1 discusses performance tuning and throughput optimization.
• Module 8.1 explains monitoring task and worker metrics.

Further Reading

• Kafka Connect Task Assignment
• Scaling Kafka Connect (Confluent Blog)

End of Module 3.3