WarpStream

Kafka-Compatible Message Queue

Synopsis

The WarpStream target writes log messages to WarpStream's serverless Kafka-compatible platform with object storage backend. WarpStream provides Kafka API compatibility with zero-disk architecture, automatic scaling, and consumption-based pricing. Configuration follows Apache Kafka patterns with WarpStream-specific endpoints and authentication.

Schema

- name: <string>
  description: <string>
  type: warpstream
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    address: <string>
    port: <integer>
    client_id: <string>
    topic: <string>
    algorithm: <string>
    username: <string>
    password: <string>
    compression: <string>
    compression_level: <string>
    acknowledgments: <string>
    allow_auto_topic_creation: <boolean>
    disable_idempotent_write: <boolean>
    max_bytes: <integer>
    max_events: <integer>
    batch_mode: <string>
    batch_separator: <string>
    tls:
      status: <boolean>
      insecure_skip_verify: <boolean>
      min_tls_version: <string>
      max_tls_version: <string>
      cert_name: <string>
      key_name: <string>
      passphrase: <string>
    field_format: <string>
    debug:
      status: <boolean>
      dont_send_logs: <boolean>

Configuration

The following fields are used to define the target:

Field	Required	Default	Description
name	Y		Target name
description	N	-	Optional description
type	Y		Must be warpstream
pipelines	N	-	Optional post-processor pipelines
status	N	true	Enable/disable the target

WarpStream Connection

Field	Required	Default	Description
address	Y	-	WarpStream broker address (provided in console). Supports a single address or a comma-separated list for multiple seed brokers. Each entry may include a port; if omitted, the port field value is appended.
port	N	9092	WarpStream broker port
client_id	N	-	Client identifier for connection tracking
topic	Y	-	Kafka topic name for message delivery

Authentication

Field	Required	Default	Description
algorithm	N	"none"	Authentication mechanism: none, plain, scram-sha-256, scram-sha-512. WarpStream requires plain, so set it explicitly
username	Y	-	WarpStream API key
password	Y	-	WarpStream API secret

Producer Settings

Field	Required	Default	Description
compression	N	"none"	Message compression: none, gzip, snappy, lz4, zstd
compression_level	N	-	Compression level (algorithm-specific)
acknowledgments	N	"leader"	Acknowledgment level: none, leader, all. none and leader are silently upgraded to all unless disable_idempotent_write: true is also set — the idempotent producer is on by default and requires all
allow_auto_topic_creation	N	false	Allow automatic topic creation if topic doesn't exist
disable_idempotent_write	N	false	Turn off the idempotent producer. Required if you want acknowledgments to actually take effect as leader or none; leaving it false keeps exactly-once-per-partition write semantics

Batch Configuration

Field	Required	Default	Description
max_bytes	N	1048576	Maximum batch size in bytes (max: 104857600). 0 does not remove the ceiling: the largest message the producer sends is then 1000012 bytes, the Kafka default maximum message size
max_events	N	1000	Maximum number of events per batch
batch_mode	N	individual	Output format: individual (one message per event), json_array (array of JSON objects), jsonl (JSON Lines with separator)
batch_separator	N	,	Separator between messages when using jsonl batch mode
field_format	N	-	Data normalization format. See applicable Normalization section

TLS Configuration

Field	Required	Default	Description
tls.status	N	false	Enable TLS/SSL encryption
tls.insecure_skip_verify	N	false	Skip certificate verification
tls.min_tls_version	N	tls1.2	Minimum TLS version: tls1.0, tls1.1, tls1.2, tls1.3
tls.max_tls_version	N	tls1.3	Maximum TLS version: tls1.0, tls1.1, tls1.2, tls1.3
tls.ca_name	N	-	CA bundle used to verify the broker certificate. When unset the host trust store is used; when set it replaces it
tls.cert_name	N	"cert.pem"	Client certificate file name for mTLS
tls.key_name	N	"key.pem"	Private key file name for mTLS
tls.passphrase	N	-	Passphrase for encrypted private key

Scheduling

See Scheduling and Pool Behavior for interval and cron fields shared by all targets.

Debug Options

Field	Required	Default	Description
debug.status	N	false	Enable debug logging
debug.dont_send_logs	N	false	Process logs but don't send to target (testing)

note

Batches are sent when either max_bytes or max_events limit is reached, whichever comes first.

Details

Authentication

API Key Authentication:

• Obtain API key and secret from WarpStream console
• Use username field for API key
• Use password field for API secret
• Authentication required for all WarpStream connections

SASL/PLAIN:

• WarpStream uses SASL/PLAIN authentication mechanism
• Set algorithm: plain — WarpStream requires it, and the target default is none
• Credentials transmitted over TLS-encrypted connection

WarpStream Credentials

Generate API keys in WarpStream console. Each key has specific permissions for topics. Ensure API key has write permissions for configured topics.

Connection Endpoints

Broker Addresses:

• WarpStream provides broker addresses in console
• Format: <cluster-name>.warpstream.com
• Default port: 9092 (Kafka protocol)
• Use provided endpoint exactly as shown in console

Multi-Region Deployment:

• WarpStream supports multi-region clusters
• Connect to regional endpoints for optimal latency
• Cross-region replication handled automatically

Topic Management

Topic Creation:

• Configure allow_auto_topic_creation: true for automatic topic creation
• Pre-create topics in WarpStream console for production use
• Topic configuration managed through WarpStream console

Topic Permissions:

• API keys grant topic-level permissions
• Verify API key has write access to configured topics
• Permission errors result in publish failures

Performance Optimization

Batch Configuration:

• Larger batches improve throughput and reduce costs
• Balance batch size against latency requirements
• WarpStream optimizes object storage writes internally

Compression:

• Enable compression to reduce bandwidth and storage costs
• Recommended algorithms: zstd (best compression), snappy (fast)
• Compression reduces WarpStream consumption-based charges

Connection Pooling:

• Maintains persistent connection to WarpStream brokers
• Automatic reconnection on connection loss
• Configurable client ID for connection tracking

Cost Optimization

WarpStream charges based on data volume processed. Enable compression and tune batch sizes to optimize costs while meeting latency requirements.

Kafka API Compatibility

Supported Features:

• Kafka Producer API
• Idempotent writes
• Batch compression
• SASL authentication
• TLS encryption

Differences from Apache Kafka:

• No ZooKeeper dependency
• Object storage backend instead of local disks
• Serverless scaling without brokers
• Different performance characteristics

Examples

Basic Configuration

Sending logs to WarpStream using API key authentication...

targets: - name: warpstream-logs type: warpstream properties: address: my-cluster.warpstream.com topic: application-logs algorithm: plain username: "${WARPSTREAM_API_KEY}" password: "${WARPSTREAM_API_SECRET}"

With Compression

Enabling Zstd compression for optimal cost and bandwidth efficiency...

targets: - name: warpstream-compressed type: warpstream properties: address: my-cluster.warpstream.com topic: telemetry-events algorithm: plain username: "${WARPSTREAM_API_KEY}" password: "${WARPSTREAM_API_SECRET}" compression: zstd compression_level: "3"

High-Volume Configuration

Optimizing for high-volume ingestion with larger batches and compression...

targets: - name: warpstream-high-volume type: warpstream properties: address: my-cluster.warpstream.com topic: metrics-stream algorithm: plain username: "${WARPSTREAM_API_KEY}" password: "${WARPSTREAM_API_SECRET}" compression: snappy max_events: 1000 max_bytes: 1048576 acknowledgments: all

With Auto Topic Creation

Allowing automatic topic creation for dynamic topic names...

targets: - name: warpstream-auto-topic type: warpstream properties: address: my-cluster.warpstream.com topic: dynamic-logs algorithm: plain username: "${WARPSTREAM_API_KEY}" password: "${WARPSTREAM_API_SECRET}" allow_auto_topic_creation: true

With TLS Encryption

Enabling TLS encryption for secure data transmission...

targets: - name: warpstream-secure type: warpstream properties: address: my-cluster.warpstream.com topic: security-logs algorithm: plain username: "${WARPSTREAM_API_KEY}" password: "${WARPSTREAM_API_SECRET}" tls: status: true min_tls_version: tls1.2 max_tls_version: tls1.3

With Normalization

Applying ECS normalization before sending to WarpStream...

targets: - name: warpstream-normalized type: warpstream properties: address: my-cluster.warpstream.com topic: normalized-events algorithm: plain username: "${WARPSTREAM_API_KEY}" password: "${WARPSTREAM_API_SECRET}" field_format: ECS compression: zstd

Production Configuration

Production-ready configuration with compression, batching, TLS, and acknowledgments...

targets: - name: warpstream-production type: warpstream properties: address: production-cluster.warpstream.com port: 9092 client_id: datastream-director-01 topic: production-telemetry algorithm: plain username: "${WARPSTREAM_API_KEY}" password: "${WARPSTREAM_API_SECRET}" compression: zstd compression_level: "3" acknowledgments: all max_events: 1000 max_bytes: 1048576 field_format: ASIM tls: status: true min_tls_version: tls1.2

Troubleshooting

The warpstream target behaves the same way as the kafka target. It opens the connection the same way, produces the same batches, and reports the same errors for the same causes. Use the Apache Kafka troubleshooting section for the full list of errors, causes and fixes. See Target Delivery Errors for how Director logs and retries target failures.

What differs for WarpStream

There are no per-broker defaults. algorithm starts at none, port at 9092 and tls.status at false, whatever the target type is. A target built from the WarpStream console details alone is therefore refused at startup with is SASL missing? or is TLS missing?. Set all three yourself.

• algorithm has to match what your cluster is configured for. WarpStream Cloud uses SASL/PLAIN, so set plain there. A mechanism the cluster does not offer is refused with UNSUPPORTED_SASL_MECHANISM even when the credential is correct.
• username is the API key and password is the API secret, both from the WarpStream console. A key that was deleted or rotated gives SASL_AUTHENTICATION_FAILED.
• tls.status has to match the listener you connect to. Set it to true whenever the cluster accepts encrypted connections, because plain on a plaintext listener sends the API secret in clear.
• port is 9092 by default, which is also WarpStream's default, so it is the one of the three you can usually leave alone. Set it if your cluster listens elsewhere.

port: 9092
algorithm: "plain"
username: "$secret{id=12}"
password: "$secret{id=13}"
tls:
  status: true

The API key carries the topic permission

WarpStream issues API keys with permissions per topic, so the right to write is a property of the key rather than a separate grant on a user. One key can therefore work for one topic and fail on the next.

TOPIC_AUTHORIZATION_FAILED on every batch is what that looks like. The target starts and reports connected, because the check it makes at startup does not mention your topic, and then every batch fails against the topic named in topic.

tip

Issue a key that covers the topic in the WarpStream console, or extend the one you have, then save the target so it reconnects. A target re-pointed at a different topic needs the permission again.

Nothing is lost while this lasts. The batch is discarded from memory, the payloads stay queued, and they are redelivered once the key carries the permission, so expect the queue to grow meanwhile.