Skip to main content

Amazon MSK

Amazon AWS Message Queue

Synopsis​

Creates a target that writes log messages to Amazon Managed Streaming for Apache Kafka (MSK) topics with support for batching, compression, and SASL or mutual TLS authentication. The target handles message delivery efficiently with configurable batch limits based on size or event count. Amazon MSK is a fully managed Apache Kafka service on AWS.

Schema​

- name: <string>
  description: <string>
  type: amazonmsk
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    address: <string>
    port: <numeric>
    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: <numeric>
    max_events: <numeric>
    batch_mode: <string>
    batch_separator: <string>
    field_format: <string>
    tls:
      status: <boolean>
      insecure_skip_verify: <boolean>
      min_tls_version: <string>
      max_tls_version: <string>
      cert_name: <string>
      key_name: <string>
      passphrase: <string>
    debug:
      status: <boolean>
      dont_send_logs: <boolean>

Configuration​

The following fields are used to define the target:

FieldRequiredDefaultDescription
nameYTarget name
descriptionN-Optional description
typeYMust be amazonmsk
pipelinesN-Optional post-processor pipelines
statusNtrueEnable/disable the target

Amazon MSK Connection​

FieldRequiredDefaultDescription
addressY-MSK broker bootstrap server address. Supports a single address or a comma-separated list for multiple seed brokers (e.g., "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com:9092,b-2.mycluster.abc123.kafka.us-east-1.amazonaws.com:9092"). Each entry may include a port; if omitted, the port field value is appended.
portN9092MSK broker port. Each listener on the cluster has its own port, so take the value from the Amazon MSK console and set it explicitly
client_idN-Client identifier for connection tracking
topicY-Kafka topic name for message delivery

Authentication​

FieldRequiredDefaultDescription
algorithmN"none"Authentication mechanism: none, plain, scram-sha-256, scram-sha-512
usernameN*-Username for SASL authentication
passwordN*-Password for SASL authentication

* = Conditionally required when using SASL authentication.

warning

IAM authentication is not available in this target. Use SASL/SCRAM, by setting algorithm to scram-sha-512 or scram-sha-256, or mutual TLS, by setting tls.cert_name and tls.key_name. A cluster that accepts only IAM cannot be reached.

Producer Settings​

FieldRequiredDefaultDescription
compressionN"none"Message compression: none, gzip, snappy, lz4, zstd
compression_levelN-Compression level (algorithm-specific)
acknowledgmentsN"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_creationNfalseAllow automatic topic creation if topic doesn't exist
disable_idempotent_writeNfalseTurn 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​

FieldRequiredDefaultDescription
max_bytesN1048576Maximum batch size in bytes (0 = unlimited, max: 104857600)
max_eventsN1000Maximum number of events per batch
batch_modeNindividualOutput format: individual (one record per message), json_array (array of JSON objects), jsonl (JSON lines with separator)
batch_separatorN","Separator between messages when using jsonl batch mode. The console fills a real newline on a new jsonl target. The separator is written exactly as stored, never unescaped, so a two-character \n is sent as a backslash and an n
field_formatN-Data normalization format. See applicable Normalization section
note

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

TLS Configuration​

FieldRequiredDefaultDescription
tls.statusNfalseEnable TLS/SSL encryption
tls.insecure_skip_verifyNfalseSkip certificate verification (not recommended)
tls.min_tls_versionNtls1.2Minimum TLS version: tls1.0, tls1.1, tls1.2, tls1.3
tls.max_tls_versionNtls1.3Maximum TLS version: tls1.0, tls1.1, tls1.2, tls1.3
tls.ca_nameN-CA bundle used to verify the broker certificate. When unset the host trust store is used; when set it replaces it
tls.cert_nameN"cert.pem"Client certificate file name for mTLS
tls.key_nameN"key.pem"Private key file name for mTLS
tls.passphraseN-Passphrase for encrypted private key

Scheduling​

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

Debug Options​

FieldRequiredDefaultDescription
debug.statusNfalseEnable debug logging
debug.dont_send_logsNfalseProcess logs but don't send to target (testing)

Details​

Amazon MSK is a fully managed Apache Kafka service on AWS. This target type allows you to connect to MSK clusters using the standard Kafka protocol.

Authentication Methods​

Two of the authentication methods an MSK cluster can offer are available here.

SASL/SCRAM Authentication

  • User name and password, held for the cluster in AWS Secrets Manager
  • Set algorithm to scram-sha-512 or scram-sha-256, with username and password
  • Set tls.status: true, since the SASL/SCRAM listener uses TLS

Mutual TLS Authentication

  • A client certificate issued by the private certificate authority the cluster trusts
  • Leave algorithm at none and set tls.status: true, tls.cert_name and tls.key_name

Unauthenticated Access (plaintext)

  • No credential is sent, and the traffic is not encrypted
  • Set algorithm: "none" and leave tls.status at false
  • Not recommended for production

IAM authentication is not available in this target, so a cluster or a listener that accepts only IAM cannot be used. Take the port for the listener you choose from the Amazon MSK console, and set port explicitly.

Connection Requirements​

MSK cluster bootstrap servers are available in the Amazon MSK Console. The address format is typically:

b-1.clustername.xxxxxx.kafka.region.amazonaws.com

For multi-broker clusters, you can use any broker from the bootstrap server list.

TLS Configuration​

An MSK cluster exposes its plaintext, TLS and SASL/SCRAM listeners on separate ports, which are listed for your cluster in the Amazon MSK console.

Set tls.status: true for every listener except the plaintext one, and add tls.cert_name and tls.key_name when the cluster asks each client for a certificate.

Message Delivery Guarantees​

The acknowledgments setting controls delivery guarantees:

LevelBehaviorUse Case
noneNo acknowledgment from brokerMaximum throughput, lowest durability
leaderAcknowledgment from partition leader onlyBalanced throughput and durability
allAcknowledgment from all in-sync replicasMaximum durability

Compression​

Message compression reduces network bandwidth and storage requirements:

AlgorithmCompression RatioCPU UsageSpeed
noneNoneMinimalFastest
gzipHighHighSlow
snappyMediumLowFast
lz4MediumLowFast
zstdHighMediumMedium

Examples​

Basic Configuration (Unauthenticated)​

The minimum configuration for an MSK target without authentication:

targets:
  - name: basic_msk
    type: amazonmsk
    properties:
      address: "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com"
      port: 9092
      topic: "application-logs"
      algorithm: "none"

With SASL/SCRAM-SHA-512​

Configuration using SASL/SCRAM-SHA-512 authentication:

targets:
  - name: msk_sasl
    type: amazonmsk
    properties:
      address: "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com"
      port: 9094
      topic: "authenticated-logs"
      algorithm: "scram-sha-512"
      username: "kafka-producer"
      password: "stored-in-secrets-manager"
      tls:
        status: true

With SASL/SCRAM-SHA-256​

Configuration using SASL/SCRAM-SHA-256 authentication:

targets:
  - name: msk_scram256
    type: amazonmsk
    properties:
      address: "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com"
      port: 9094
      topic: "secure-logs"
      algorithm: "scram-sha-256"
      username: "kafka-user"
      password: "secure-password"
      compression: "snappy"
      tls:
        status: true

With Mutual TLS​

Configuration for a cluster that authenticates clients by certificate:

targets:
  - name: msk_mtls
    type: amazonmsk
    properties:
      address: "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com"
      port: 9094
      topic: "authenticated-logs"
      algorithm: "none"
      tls:
        status: true
        cert_name: "msk-client.pem"
        key_name: "msk-client-key.pem"

With Compression​

Configuration with zstd compression:

targets:
  - name: compressed_msk
    type: amazonmsk
    properties:
      address: "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com"
      port: 9094
      topic: "compressed-logs"
      algorithm: "scram-sha-256"
      username: "kafka-producer"
      password: "password-here"
      compression: "zstd"
      tls:
        status: true

High Throughput​

Configuration optimized for maximum throughput:

targets:
  - name: high_throughput_msk
    type: amazonmsk
    properties:
      address: "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com"
      port: 9094
      topic: "high-volume-logs"
      algorithm: "scram-sha-512"
      username: "kafka-producer"
      password: "password-here"
      compression: "lz4"
      acknowledgments: "leader"
      max_bytes: 1048576
      max_events: 10000
      tls:
        status: true

High Reliability​

Configuration optimized for maximum durability:

targets:
  - name: reliable_msk
    type: amazonmsk
    pipelines:
      - checkpoint
    properties:
      address: "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com"
      port: 9094
      topic: "critical-logs"
      algorithm: "scram-sha-512"
      username: "kafka-producer"
      password: "password-here"
      compression: "zstd"
      acknowledgments: "all"
      max_events: 100
      disable_idempotent_write: false
      tls:
        status: true
        min_tls_version: "tls1.3"

With Field Normalization​

Using field normalization for standard format:

targets:
  - name: normalized_msk
    type: amazonmsk
    properties:
      address: "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com"
      port: 9094
      topic: "normalized-logs"
      algorithm: "scram-sha-256"
      username: "kafka-producer"
      password: "password-here"
      field_format: "cim"
      compression: "snappy"
      tls:
        status: true

With Client ID​

Configuration with client ID for tracking:

targets:
  - name: msk_tracked
    type: amazonmsk
    properties:
      address: "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com"
      port: 9094
      topic: "tracked-logs"
      client_id: "datastream-producer-01"
      algorithm: "scram-sha-256"
      username: "kafka-producer"
      password: "password-here"
      compression: "lz4"
      tls:
        status: true

Scheduled Batching​

Configuration with scheduled batch delivery:

targets:
  - name: scheduled_msk
    type: amazonmsk
    properties:
      address: "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com"
      port: 9094
      topic: "scheduled-logs"
      algorithm: "scram-sha-512"
      username: "kafka-producer"
      password: "password-here"
      max_events: 5000
      interval: "5m"
      compression: "gzip"
      tls:
        status: true

Auto Topic Creation​

Configuration with automatic topic creation enabled:

targets:
  - name: auto_topic_msk
    type: amazonmsk
    properties:
      address: "b-1.mycluster.abc123.kafka.us-east-1.amazonaws.com"
      port: 9094
      topic: "dynamic-logs"
      algorithm: "scram-sha-256"
      username: "kafka-producer"
      password: "password-here"
      allow_auto_topic_creation: true
      compression: "snappy"
      max_events: 500
      tls:
        status: true

Troubleshooting​

The amazonmsk target connects, authenticates and produces in exactly the same way as the kafka target, so it reports the same errors for the same causes: authentication and encryption failures at startup, topic authorization and missing topic failures on the first batch, and the size limits on a single record and on a whole batch. Each error is listed with its cause and its fix on Apache Kafka. Log lines and the connection status carry this target's name, so match on the cause text, which is the part after Reason: or after the last colon, rather than on the target name in the examples there. See Target Delivery Errors for how Director logs and retries target failures.

What differs for Amazon MSK​

Three points decide whether a target for MSK can connect at all. Nothing in the target is preset for this service.

  • algorithm, port and tls.status all start at their generic defaults, which are none, 9092 and false. Nothing about the type amazonmsk changes them. A target built from the bootstrap address alone therefore offers no credential and no encryption, and it fails at startup against an authenticated listener with is SASL missing?. Set all three to match the listener you connect to, and take the port from the Amazon MSK console, because each listener on the cluster has its own.

  • IAM authentication is not available in this target. A cluster or a listener that accepts only IAM cannot be used, whatever AWS credentials the Director host holds. A target pointed at the IAM listener fails at startup with is SASL missing? or UNSUPPORTED_SASL_MECHANISM, and an IAM value in algorithm is refused with algorithm must be plain, scram-sha-256, scram-sha-512 or gssapi.

  • Use SASL/SCRAM or mutual TLS instead. For SASL/SCRAM, set algorithm to scram-sha-512 or scram-sha-256, set username and password to the SCRAM user held for the cluster in AWS Secrets Manager, and set tls.status: true. For mutual TLS, leave algorithm at none and set tls.status: true together with tls.cert_name and tls.key_name. Both mechanisms need the listener that offers them, so change the port when you change the mechanism.

note

The right to write to the topic is granted separately, as an ACL on the principal you authenticate as. It is not tested at startup, so a target with a valid credential and no write ACL reports connected and then fails every batch with TOPIC_AUTHORIZATION_FAILED.

Every broker has to be reachable, not just the bootstrap address​

The addresses in address are used only to find the cluster. The brokers answer with their own addresses, and batches go to those, so each broker has to resolve and be reachable from the Director host. When only the bootstrap address is reachable, the target starts cleanly, reports connected, and then waits with a growing queue and nothing in the log.

A path that is blocked rather than closed shows at startup as dial tcp ...: i/o timeout, typically a security group or a route that does not allow the port from the Director host. The connection is plain TCP and ignores HTTP_PROXY, so an egress path that offers only a proxy does not work.

The Amazon MSK console lists the bootstrap addresses for the cluster, with a separate set for each authentication type it offers. Take the set that matches the mechanism you configured, and put every broker in it into address as a comma-separated list, so the target can still start when one broker is down.