5.1 – Performance Tuning & Throughput Optimization

Overview

This module is the authoritative guide to performance tuning and throughput optimization for Kafka Connect. By the end, you will understand:

• Key performance configurations for connectors and tasks.
• Batch size, poll intervals, and throughput tuning.
• How task scaling affects performance.
• Converter performance impact.
• Backpressure behavior and how to handle it.

All performance tuning concepts live here. Other modules will reference this when discussing specific connectors or operational scenarios, but will not re-explain these concepts.

Performance Fundamentals

What Determines Throughput?

Kafka Connect throughput is determined by:

1. Task Parallelism: More tasks = higher throughput (up to external system limits).
2. Batch Size: Larger batches = fewer round-trips, higher throughput (but higher latency).
3. Poll Intervals: How frequently tasks poll for data.
4. Converter Overhead: Serialization/deserialization cost (Avro vs. JSON vs. String).
5. External System Performance: Database query speed, API rate limits, network latency.
6. Worker Resources: CPU, memory, network bandwidth.

Key Insight: Kafka Connect throughput is often bottlenecked by the external system, not Kafka itself. Tuning Connect configs without optimizing the external system yields diminishing returns.

Source Connector Performance Tuning

1. Polling Frequency

Configuration: poll.interval.ms

Purpose: How often the source task polls the external system for new data.

Default: 5000 (5 seconds)

Trade-off:

• Lower values (e.g., 1000): Lower latency, higher load on external system.
• Higher values (e.g., 10000): Higher latency, lower load on external system.

Example:

{
  "poll.interval.ms": "1000"
}

Best Practice: Start with the default, then decrease if low latency is critical and the external system can handle the load.

2. Batch Size (Source-Specific)

Configuration: Varies by connector (e.g., batch.size for JDBC Source, batch.max.rows for some connectors).

Purpose: Number of records read per poll from the external system.

Default: Connector-specific (e.g., JDBC Source defaults to 100).

Trade-off:

• Larger batches: Higher throughput, fewer queries to external system, but higher memory usage.
• Smaller batches: Lower memory, more frequent queries, lower throughput.

Example (JDBC Source):

{
  "batch.max.rows": "1000"
}

Best Practice: Increase batch size if the external system can handle large result sets without timing out.

3. Task Parallelism

Configuration: tasks.max

Purpose: Number of tasks the connector creates for parallel execution.

Impact: More tasks = more concurrent queries to the external system.

Example:

{
  "tasks.max": "10"
}

Best Practice:

• For JDBC Source: Set tasks.max to the number of tables (one task per table).
• For File Source: Set to the number of files.
• Monitor external system load to avoid overwhelming it.

See Module 3.3 for detailed scaling strategies.

4. Connection Pooling (JDBC Source)

Configuration: Connector-specific (e.g., connection.pool.size).

Purpose: Number of database connections each task maintains.

Default: Varies by connector.

Trade-off:

• Larger pool: More concurrent queries, higher throughput, but more database load.
• Smaller pool: Lower database load, but tasks may wait for available connections.

Example:

{
  "connection.pool.size": "10"
}

Best Practice: Match pool size to tasks.max (e.g., if tasks.max=5, set pool size to 5-10).

5. Query Optimization (JDBC Source)

Configuration: query or mode (incrementing, timestamp, bulk).

Purpose: How the connector queries the database.

Modes:

• Incrementing: SELECT * FROM table WHERE id > last_offset (efficient for append-only tables).
• Timestamp: SELECT * FROM table WHERE updated_at > last_timestamp (requires indexed timestamp column).
• Bulk: SELECT * FROM table (full table scan, inefficient for large tables).

Best Practice:

• Use incrementing or timestamp mode with indexed columns.
• Avoid bulk mode for large tables (millions of rows).
• Add database indexes on incrementing.column.name or timestamp.column.name.

Example:

{
  "mode": "timestamp",
  "timestamp.column.name": "updated_at",
  "query": "SELECT * FROM users WHERE updated_at > ? AND deleted = false"
}

Sink Connector Performance Tuning

1. Consumer Batch Size

Configuration: consumer.max.poll.records

Purpose: Maximum number of records fetched per poll from Kafka.

Default: 500

Trade-off:

• Larger batches: Higher throughput, fewer poll requests to Kafka, but higher memory usage.
• Smaller batches: Lower memory, more frequent polls, lower throughput.

Example:

{
  "consumer.max.poll.records": "3000"
}

Best Practice: Increase if the external system can handle large batch writes (e.g., database bulk inserts, Elasticsearch bulk indexing).

2. Flush Frequency

Configuration: flush.size (connector-specific)

Purpose: Number of records buffered before writing to the external system.

Default: Varies by connector (e.g., Elasticsearch Sink defaults to 500).

Trade-off:

• Larger flush size: Higher throughput, fewer write operations, but higher memory usage and latency.
• Smaller flush size: Lower latency, more frequent writes, lower throughput.

Example (Elasticsearch Sink):

{
  "batch.size": "1000",
  "max.buffered.records": "10000"
}

Best Practice: Match batch.size to the external system's bulk write limits (e.g., Elasticsearch supports up to 10,000 documents per bulk request).

3. Task Parallelism

Configuration: tasks.max

Purpose: Number of tasks writing to the external system concurrently.

Impact: More tasks = more concurrent writes.

Example:

{
  "tasks.max": "6"
}

Best Practice:

• Set tasks.max to the number of Kafka partitions (or less).
• For JDBC Sink: Monitor database connection pool usage.
• For Elasticsearch Sink: Monitor indexing throughput and throttling.

See Module 3.3 for detailed scaling strategies.

4. Write Strategy (JDBC Sink)

Configuration: insert.mode

Purpose: How records are written to the database.

Modes:

• Insert: INSERT statements (fast, but fails on duplicate keys).
• Upsert: INSERT ... ON CONFLICT UPDATE (slower, but idempotent).
• Update: UPDATE only (requires existing rows).

Trade-off:

• Insert: Highest throughput, but not idempotent (duplicates cause errors).
• Upsert: Idempotent, but 20-30% slower than insert.

Example:

{
  "insert.mode": "upsert",
  "pk.mode": "record_key",
  "pk.fields": "user_id"
}

Best Practice: Use upsert for at-least-once delivery guarantees (prevents duplicates).

5. Indexing Strategy (Elasticsearch Sink)

Configuration: linger.ms (buffer time before flushing).

Purpose: How long to wait before flushing buffered records.

Default: 0 (flush immediately after batch is full).

Trade-off:

• Higher linger: Better batching, higher throughput, but higher latency.
• Lower linger: Lower latency, but smaller batches.

Example:

{
  "linger.ms": "100"
}

Best Practice: Set linger.ms to 50-200ms for high-throughput use cases.

Converter Performance Impact

Serialization/Deserialization Overhead

Converters add CPU overhead for serialization (source) and deserialization (sink).

Performance Comparison (approximate):

Best Practice:

• Use Avro or Protobuf for high-throughput, schema-aware pipelines.
• Use JSON (no schema) for simplicity if schema evolution isn't needed.
• Avoid JSON (with schema) for high-throughput use cases (large message size).

Example: Switch from JSON to Avro

Before (JSON with schema):

{
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "true"
}

After (Avro):

{
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "http://localhost:8081"
}

Impact: 30-50% throughput increase + 40-60% message size reduction.

Backpressure Behavior

What is Backpressure?

Backpressure occurs when the external system (or Kafka) cannot keep up with the rate of data being produced/consumed, causing Connect tasks to slow down.

Source Connector Backpressure

Cause: Kafka brokers are slow to acknowledge writes (e.g., high broker load, replication lag).

Effect: Source tasks block on producer.send(), slowing poll loops.

Solution:

1. Increase broker resources (CPU, disk I/O).
2. Increase buffer.memory (default: 32 MB) to buffer more records.
3. Tune linger.ms (default: 0) to batch writes more effectively.

Example:

{
  "producer.buffer.memory": "67108864",
  "producer.linger.ms": "10"
}

Sink Connector Backpressure

Cause: External system is slow to process writes (e.g., database throttling, Elasticsearch indexing bottleneck).

Effect: Sink tasks slow down, consumer lag increases.

Solution:

1. Scale external system (add database replicas, scale Elasticsearch cluster).
2. Increase tasks.max to parallelize writes.
3. Tune batch sizes (consumer.max.poll.records, batch.size).

Example:

{
  "consumer.max.poll.records": "5000",
  "batch.size": "1000"
}

Worker-Level Performance Tuning

1. JVM Heap Size

Configuration: KAFKA_HEAP_OPTS environment variable.

Default: -Xmx1G (1 GB)

Recommendation:

• Low-throughput connectors (< 10K records/sec): 1-2 GB.
• High-throughput connectors (> 100K records/sec): 4-8 GB.

Example:

export KAFKA_HEAP_OPTS="-Xmx4G -Xms4G"

2. Garbage Collection (GC) Tuning

Configuration: KAFKA_JVM_PERFORMANCE_OPTS

Default: G1GC (good for most workloads).

Recommendation: Use G1GC for low-latency, Parallel GC for high-throughput batch jobs.

Example:

export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC -XX:MaxGCPauseMillis=20"

3. Network Threads

Configuration: None (JVM default).

Recommendation: Ensure sufficient network bandwidth for high-throughput connectors (10 Gbps+ for large-scale deployments).

Scaling Throughput: A Practical Example

Scenario: Ingest 1 Million Records/Minute from PostgreSQL

Initial Setup:

• 1 JDBC Source Connector
• tasks.max=1
• Default batch size (100)
• Default poll interval (5000ms)

Measured Throughput: 5,000 records/minute (far below target).

Optimization Steps

Step 1: Increase Batch Size

{
  "batch.max.rows": "5000"
}

Result: 50,000 records/minute (10x improvement).

Step 2: Increase Task Parallelism

{
  "tasks.max": "10"
}

Result: 500,000 records/minute (100x improvement).

Step 3: Optimize Database Query

Add index on id column:

CREATE INDEX idx_users_id ON users(id);

Result: 800,000 records/minute (160x improvement).

Step 4: Increase Worker Resources

Scale from 3 workers (2 CPU, 4 GB RAM each) to 5 workers (4 CPU, 8 GB RAM each).

Result: 1,200,000 records/minute (240x improvement, exceeds target).

Performance Monitoring

Key Metrics to Monitor

Source Connector Metrics (JMX)

Sink Connector Metrics (JMX)

Worker Metrics (JMX)

Monitoring Tools

• JMX Exporter + Prometheus + Grafana: Industry-standard observability stack.
• Confluent Control Center: Commercial UI for monitoring Connect clusters.
• Kafka Connect REST API: /connectors/{name}/status for real-time status.

See Module 8.1 for detailed monitoring strategies.

Key Takeaways

1. Throughput is determined by task parallelism, batch size, poll intervals, converters, and external system performance.
2. Source connectors: Tune poll.interval.ms, batch.max.rows, tasks.max, and database queries.
3. Sink connectors: Tune consumer.max.poll.records, batch.size, tasks.max, and write strategies.
4. Converters: Avro/Protobuf > JSON (no schema) > JSON (with schema) for throughput.
5. Backpressure: Monitor and scale external systems, not just Connect.
6. Worker resources: Scale JVM heap, workers, and network bandwidth for high-throughput deployments.
7. Iterative tuning: Start with defaults, measure, optimize one parameter at a time.

What's Next?

• Module 6.1 covers error handling, retries, and Dead Letter Queues.
• Module 8.1 explains monitoring and metrics in detail.
• Module 3.3 provides detailed task scaling strategies.

Further Reading

• Kafka Connect Performance Tuning (Confluent Blog)
• JDBC Source Connector Performance

End of Module 5.1