Kafka

The Parseable Kafka connector lets ingestors read JSON records from Kafka topics and store them as Parseable datasets. Use it when Kafka is already part of your data path and you want Parseable to consume from the topic directly instead of adding another shipper in between.

Features

The connector supports Kafka consumer configuration, configurable per-partition buffering, TLS and SASL authentication, consumer-group based scaling, and partition-aware processing. Producer settings are documented for future dead-letter-topic support, but normal ingestion only requires the consumer path.

Configuration Options

General Kafka Configuration

All parameters can be set using command-line arguments or environment variables. Environment variables take precedence over default values. For ingestion, start with the general and consumer configuration tables, then add security settings based on your Kafka cluster.

For more details, refer to Kafka's official documentation.

Consumer Configuration

• All parameters can be set using command-line arguments or environment variables.
• Environment variables take precedence over default values.
• Some values, such as group_instance_id, are dynamically generated if not explicitly provided.

For more details, refer to Kafka's official documentation on consumer configurations.

Producer Configuration

• All parameters can be set using command-line arguments or environment variables.
• Environment variables take precedence over default values.
• Certain parameters, such as --producer-acks and --producer-enable-idempotence, affect message durability and reliability.

For more details, refer to Kafka's official documentation on producer configurations.

Security Configuration

Security Configuration Combinations

Plaintext Communication (No Security)

• --security-protocol=PLAINTEXT
• No additional parameters required.

SSL Encryption

• --security-protocol=SSL

Required parameters:

• --ssl-ca-location
• --ssl-certificate-location
• --ssl-key-location
• --ssl-key-password (if the key is password-protected)

SASL Authentication with SSL

• --security-protocol=SASL_SSL

Required parameters:

• --sasl-mechanism
• --sasl-username and --sasl-password (for PLAIN or SCRAM mechanisms)
• --kerberos-service-name and --kerberos-principal (for GSSAPI mechanism)
• --kerberos-keytab (if using Kerberos authentication)

SSL parameters (if required by the security policy)

SASL Authentication without SSL

• --security-protocol=SASL_PLAINTEXT

Required parameters:

• --sasl-mechanism
• --sasl-username and --sasl-password (for PLAIN or SCRAM mechanisms)
• --kerberos-service-name and --kerberos-principal (for GSSAPI mechanism)
• --kerberos-keytab (if using Kerberos authentication)

OAuth Bearer Token Authentication

• --security-protocol=SASL_SSL
• --sasl-mechanism=OAUTHBEARER

Provider-specific parameters:

• Amazon MSK IAM: --oauth-provider=aws-msk and --aws-region=<region>. AWS credentials are resolved through the AWS SDK default credential chain.
• Standard OIDC: --oauth-provider=oidc, --oauth-token-endpoint-url, --oauth-client-id, and --oauth-client-secret.

SASL_PLAINTEXT is rejected for OAuth bearer tokens because it would expose the token without TLS.

Examples

SSL Configuration

export P_KAFKA_SECURITY_PROTOCOL="SSL"
export P_KAFKA_SSL_CA_LOCATION="/path/to/ca.pem"
export P_KAFKA_SSL_CERTIFICATE_LOCATION="/path/to/client-cert.pem"
export P_KAFKA_SSL_KEY_LOCATION="/path/to/client-key.pem"
export P_KAFKA_SSL_KEY_PASSWORD="my-secure-password"

SASL Configuration

export P_KAFKA_SECURITY_PROTOCOL="SASL_SSL"
export P_KAFKA_SASL_MECHANISM="SCRAM-SHA-512"
export P_KAFKA_SASL_USERNAME="my-user"
export P_KAFKA_SASL_PASSWORD="my-password"

For more details, refer to Kafka's official security documentation.

Concurrency and Multi-Instance Processing

The connector uses a dedicated thread per partition with configurable concurrency:

export P_KAFKA_PARTITION_LISTENER_CONCURRENCY="2"

Thread Assignment Formula:

Threads per ingestor = min(Partitions per ingestor, Configured threads)

Example Scenarios:

Balanced Configuration:

• 6 partitions
• 2 ingest nodes
• 3 threads per node

Result: Each thread handles one partition

Over-threaded Configuration:

• 4 partitions
• 2 ingest nodes
• 4 threads per node

Result: 2 threads per node will be idle

Consumer Group Rebalancing

Rebalance triggers:

• New consumer joins
• Existing consumer leaves
• Network issues
• Partition reassignment

Available strategies:

• Range: Assigns consecutive partitions
• RoundRobin: Distributes evenly
• Sticky: Minimizes reassignments (Not recommended for Parseable since rdKafka issues)
• CooperativeSticky: Controlled rebalance (Not recommended for Parseable since rdKafka support is lacking)

Recommended configuration(by default):

export P_KAFKA_CONSUMER_PARTITION_STRATEGY="roundrobin,range"
export P_KAFKA_CONSUMER_SESSION_TIMEOUT="60000"
export P_KAFKA_CONSUMER_HEARTBEAT_INTERVAL="3000"

Metrics

All metrics listed in librdkafka's statistics documentation are available at the /metrics endpoint.

Architecture and Design

Core Components

• KafkaStreams: Manages consumers and partitions.
• StreamWorker: Processes records per partition.
• ParseableSinkProcessor: Transforms messages and sinks them to Parseable.
• RebalanceListener: Handles partition rebalancing.
• Metrics Collector: Provides Prometheus metrics.

Security Layer: Configurable authentication.

Error Handling

Best Practices

Performance Tuning:

• More partitions = Better parallelism
• Use RoundRobin partition assignment
• Increase partition count for high lag

Security:

• Prefer SSL over SASL_PLAINTEXT
• Rotate certificates regularly
• Monitor authentication failures