Apache Kafka

Scanner can ingest data from self-managed Apache Kafka clusters. Rather than consuming from Kafka directly, Scanner reads from an S3 collect buffer bucket that a Kafka Connect S3 sink connector writes to. This lets you reuse the Kafka Connect infrastructure you likely already run, and keeps Scanner out of your consumer group management.

Step 1: Set up the collect buffer bucket

Follow the Streams overview to create an S3 collect buffer bucket with a 7-day lifecycle policy and link it to Scanner.

Step 2: Run an S3 sink connector on Kafka Connect

We recommend the Aiven S3 sink connector for Apache Kafka. It is Apache 2.0 licensed and can natively write newline-delimited JSON with gzip compression, which is exactly the format Scanner ingests most easily. Use the 3.x releases; the older 2.x line (aiven/s3-connector-for-apache-kafka) requires static AWS keys and cannot authenticate via IAM roles or instance profiles.

Example connector configuration:

{
  "name": "scanner-collect-buffer-sink",
  "connector.class": "io.aiven.kafka.connect.s3.AivenKafkaConnectS3SinkConnector",
  "topics": "my-logs-topic",
  "aws.s3.bucket.name": "my-company-scanner-collect-buffer",
  "aws.s3.region": "us-east-1",
  "aws.s3.prefix": "kafka/my-logs-topic/{{ utc_date }}/",
  "format.output.type": "jsonl",
  "format.output.fields": "value",
  "format.output.envelope": "false",
  "file.compression.type": "gzip",
  "file.max.records": "10000",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false",
  "errors.tolerance": "all",
  "errors.log.enable": "true"
}

Key settings:

• format.output.type: jsonl writes one JSON event per line (newline-delimited JSON).
• format.output.fields: value with format.output.envelope: false writes each record value as plain JSON. format.output.envelope defaults to true, which nests every event under a {"value": ...} wrapper.
• file.compression.type: gzip compresses each output file.
• errors.tolerance: all skips records that fail conversion (e.g. non-JSON messages) instead of permanently failing the connector task; errors.log.enable: true logs each skipped record.
• aws.s3.prefix supports {{ utc_date }} for date-partitioned keys; file.name.template offers finer control. See the Aiven S3 sink documentation.
• AWS credentials come from the standard AWS credential chain (environment variables, instance profiles, etc.), or can be set with aws.access.key.id / aws.secret.access.key.

Files are flushed to S3 when file.max.records is reached or at the Connect offset flush interval (offset.flush.interval.ms, 60 seconds by default), so data lands in S3 within about a minute even on low-volume topics.

Plaintext topics

If a topic carries plaintext log lines rather than JSON, JsonConverter would fail on every message, and errors.tolerance: all would skip them all (logged, but nothing ingested). Instead set:

{
  "value.converter": "org.apache.kafka.connect.storage.StringConverter",
  "format.output.envelope": "true"
}

Each line is then written as {"value": "<log line>"}, still gzip JSONL, with the message text under the value field in Scanner.

Alternative connectors

The errors.tolerance and converter guidance above is Kafka Connect framework-level and applies to these connectors too.

• Confluent S3 Sink Connector: the most widely deployed S3 sink. Licensed under the Confluent Community License, which permits self-hosting for your own use. Use format.class of io.confluent.connect.s3.format.json.JsonFormat (writes one JSON record per line) with s3.compression.type: gzip, and set value.converter.schemas.enable: false to avoid wrapping each record in a schema/payload envelope. For low-volume topics, set rotate.schedule.interval.ms so files are flushed on wall-clock time rather than waiting for flush.size records to accumulate. For plaintext topics, use io.confluent.connect.s3.format.bytearray.ByteArrayFormat and ingest as Plaintext instead of JsonLines.
• Lenses Stream Reactor AWS S3 sink: Apache 2.0 licensed, configured with SQL-like KCQL statements, e.g. INSERT INTO my-bucket:prefix SELECT * FROM my-topic STOREAS 'JSON'.

Step 3: Ingest via Scanner Collect

Follow the Custom Logs - AWS S3 guide to ingest the collect buffer bucket via Scanner Collect, using File Type JsonLines and Compression Gzip.