Publish Events to Kafka

MinIO AIStor supports publishing bucket notification events to a Kafka service endpoint.

MinIO AIStor relies on the franz-go client library for Kafka connectivity and shares that project’s Kafka version support.

Kafka behaves differently from an HTTP endpoint when a broker fails. See Plan for broker failures before running a Kafka target in production.

Prerequisites

Kafka minimum versions and supported versions.

MinIO AIStor relies on the franz-go client library for Kafka connectivity and shares that project’s Kafka version support.
The MinIO AIStor mc command line tool.

This procedure uses the mc command line tool for certain actions. See the mc Quickstart for installation instructions.

Add a Kafka endpoint to a MinIO AIStor deployment

The following procedure adds a new Kafka service endpoint for supporting bucket notifications in a MinIO AIStor deployment.

Add the Kafka endpoint to MinIO AIStor.

You can configure a new Kafka service endpoint using either environment variables or by setting runtime configuration settings.
MinIO AIStor supports specifying the Kafka service endpoint and associated configuration settings using environment variables. The minio server process applies the specified settings on its next startup.
The following example code sets all environment variables related to configuring a Kafka service endpoint. The minimum required variables are:
- MINIO_NOTIFY_KAFKA_ENABLE
- MINIO_NOTIFY_KAFKA_BROKERS
The following lists all available variables:
export MINIO_NOTIFY_KAFKA_ENABLE_<IDENTIFIER>="on" export MINIO_NOTIFY_KAFKA_BROKERS_<IDENTIFIER>="<ENDPOINT>" export MINIO_NOTIFY_KAFKA_TOPIC_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_SASL_USERNAME_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_SASL_PASSWORD_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_SASL_MECHANISM_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_TLS_CLIENT_AUTH_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_SASL_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_TLS_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_TLS_SKIP_VERIFY_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_CLIENT_TLS_CERT_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_CLIENT_TLS_KEY_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_QUEUE_DIR_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_QUEUE_LIMIT_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_VERSION_<IDENTIFIER>="<string>" export MINIO_NOTIFY_KAFKA_COMMENT_<IDENTIFIER>="<string>"
- Replace <IDENTIFIER> with a unique descriptive string for the Kafka service endpoint. Use the same <IDENTIFIER> value for all environment variables related to the new target service endpoint. The examples in this procedure assume an identifier of primary.
  
  If the specified <IDENTIFIER> matches an existing Kafka service endpoint on the AIStor Server, the new settings override any existing settings for that endpoint. Use mc admin config get notify_kafka to review the currently configured Kafka endpoints on the AIStor Server.
- Replace <ENDPOINT> with a comma-separated list of Kafka brokers. For example:
  
  "kafka1.example.com:2021,kafka2.example.com:2021"
- Set MINIO_NOTIFY_KAFKA_SASL_MECHANISM_<IDENTIFIER> to the short name of the SASL mechanism your Kafka brokers use. AIStor Server accepts sha512 for SCRAM-SHA-512, sha256 for SCRAM-SHA-256, and gssapi for Kerberos. Any other value selects PLAIN.
  
  MINIO_NOTIFY_KAFKA_SASL_<IDENTIFIER> is a separate setting that turns SASL on. Set it to "on", and do not put the mechanism name in it. AIStor Server ignores the mechanism, the username, and the password unless this setting is on.
See Kafka Service for Bucket Notifications for complete documentation on each environment variable.
MinIO AIStor supports adding or updating Kafka endpoints on a running minio server process using the mc admin config set command and the notify_kafka configuration key. MinIO AIStor applies a new or updated configuration setting immediately, with no server restart.
The following example code sets all settings related to configuring a Kafka service endpoint. The minimum required setting is notify_kafka brokers:
mc admin config set ALIAS notify_kafka:IDENTIFIER \ brokers="<ENDPOINT>" \ topic="<string>" \ sasl="<string>" \ sasl_username="<string>" \ sasl_password="<string>" \ sasl_mechanism="<string>" \ tls_client_auth="<string>" \ tls="<string>" \ tls_skip_verify="<string>" \ client_tls_cert="<string>" \ client_tls_key="<string>" \ version="<string>" \ queue_dir="<string>" \ queue_limit="<string>" \ comment="<string>"
- Replace IDENTIFIER with a unique descriptive string for the Kafka service endpoint. The examples in this procedure assume an identifier of primary.
  
  If the specified IDENTIFIER matches an existing Kafka service endpoint on the AIStor Server, the new settings override any existing settings for that endpoint. Use mc admin config get notify_kafka to review the currently configured Kafka endpoints on the AIStor Server.
- Replace ENDPOINT with a comma separated list of Kafka brokers. For example:
  
  "kafka1.example.com:2021,kafka2.example.com:2021"
- Set sasl_mechanism to the short name of the SASL mechanism your Kafka brokers use. AIStor Server accepts sha512 for SCRAM-SHA-512, sha256 for SCRAM-SHA-256, and gssapi for Kerberos. Any other value selects PLAIN.
  
  Set sasl to "on" as well. AIStor Server ignores sasl_mechanism, sasl_username, and sasl_password unless sasl is on.
See Kafka Bucket Notification Configuration Settings for complete documentation on each setting.
If you define both environment variables and configuration settings for the deployment, the settings defined by environment variables are used.
Confirm the target is registered.

MinIO AIStor applies a configuration setting change immediately.

If you configured this endpoint with environment variables, those values take precedence and mc admin config set has no effect on it. Update the environment variables on every node, then restart the deployment with the mc admin service restart command:
```
mc admin service restart ALIAS
```
Replace ALIAS with the alias of the deployment.

List the notification targets with mc event target list:
```
mc event target list ALIAS
```
The Kafka target appears with an ARN similar to the following:
```
arn:minio:sqs::primary:kafka
```
You must specify the ARN resource when configuring bucket notifications with the associated Kafka deployment as a target.
Configure bucket notifications using the Kafka endpoint as a target.

Use the mc event add command to add a new bucket notification event with the configured Kafka service as a target:
```
mc event add ALIAS/BUCKET arn:minio:sqs::primary:kafka \
  --event EVENTS
```
- Replace ALIAS with the alias of an AIStor Server.
- Replace BUCKET with the name of the bucket in which to configure the event.
- Replace EVENTS with a comma-separated list of events for which MinIO AIStor triggers notifications.
Use mc event ls to view all configured bucket events for a given notification target:
```
mc event ls ALIAS/BUCKET arn:minio:sqs::primary:kafka
```
Validate the configured events.

Perform an action on the bucket for which you configured the new event and check the Kafka service for the notification data. The action required depends on which events were specified when configuring the bucket notification.

For example, if the bucket notification configuration includes the s3:ObjectCreated:Put event, you can use the mc cp command to create a new object in the bucket and trigger a notification.
```
mc cp ~/data/new-object.txt ALIAS/BUCKET
```

Update a Kafka endpoint in an AIStor Server

The following procedure updates an existing Kafka service endpoint for supporting bucket notifications in an AIStor Server.

List configured Kafka endpoints in the deployment.

Use the mc admin config get command to list the currently configured Kafka service endpoints in the deployment:
```
mc admin config get ALIAS notify_kafka
```
Replace ALIAS with the alias of the AIStor Server.

The command output resembles the following:
```
notify_kafka:primary tls_skip_verify="off"  queue_dir="" queue_limit="0" sasl="off" sasl_password="" sasl_username="" tls_client_auth="0" tls="off" brokers="" topic="" client_tls_cert="" client_tls_key="" version=""
notify_kafka:secondary tls_skip_verify="off"  queue_dir="" queue_limit="0" sasl="off" sasl_password="" sasl_username="" tls_client_auth="0" tls="off" brokers="" topic="" client_tls_cert="" client_tls_key="" version=""
```
- notify_kafka is the top-level configuration key for a Kafka Notification Settings.
- brokers specifies the Kafka service endpoint for the given notify_kafka key.
- The suffix <IDENTIFIER> on notify_kafka describes the unique identifier for that Kafka service endpoint.
Note the identifier for the Kafka service endpoint you want to update for the next step.

Update the Kafka endpoint.

Use the mc admin config set command to set the new configuration for the Kafka service endpoint:

mc admin config set ALIAS notify_kafka:<IDENTIFIER> \
   brokers="kafka1.example.net:9200,kafka2.example.net:9200" \
   topic="<string>" \
   sasl="<string>" \
   sasl_username="<string>" \
   sasl_password="<string>" \
   sasl_mechanism="<string>" \
   tls_client_auth="<string>" \
   tls="<string>" \
   tls_skip_verify="<string>" \
   client_tls_cert="<string>" \
   client_tls_key="<string>" \
   version="<string>" \
   queue_dir="<string>" \
   queue_limit="<string>" \
   comment="<string>"

The notify_kafka brokers configuration setting is the minimum required for a Kafka service endpoint. All other configuration settings are optional. See Kafka Notification Settings for a complete list of Kafka configuration settings.

Confirm the target is registered.

MinIO AIStor applies a configuration setting change immediately.

If you configured this endpoint with environment variables, those values take precedence and mc admin config set has no effect on it. Update the environment variables on every node, then restart the deployment with the mc admin service restart command:
```
mc admin service restart ALIAS
```
Replace ALIAS with the alias of the deployment.

List the notification targets with mc event target list:
```
mc event target list ALIAS
```
The Kafka target appears with an ARN similar to the following:
```
arn:minio:sqs::primary:kafka
```
Validate the changes.

Perform an action on a bucket which has an event configuration using the updated Kafka service endpoint and check the Kafka service for the notification data. The action required depends on which events were specified when configuring the bucket notification.

For example, if the bucket notification configuration includes the s3:ObjectCreated:Put event, you can use the mc cp command to create a new object in the bucket and trigger a notification.
```
mc cp ~/data/new-object.txt ALIAS/BUCKET
```

Plan for broker failures

A Kafka target does not fail over between brokers the way an HTTP target fails over behind a load balancer. MinIO AIStor sends each event to the broker that leads the event’s partition, and only Kafka can move that leadership. If a broker stays in the Kafka cluster but stops answering clients, MinIO AIStor waits for that broker, and delivery to the target stops on every server.

Set a queue directory

Set queue_dir for every Kafka target in production.

With a queue directory, MinIO AIStor writes each event to disk and sends it to Kafka in the background. A broker failure does not slow S3 requests, and events wait on disk until Kafka accepts them.

Without a queue directory, a stalled broker holds the workers that deliver events to every notification target, not only to Kafka. MinIO AIStor then drops events for all targets. If api sync_events is on, S3 write requests also wait for Kafka and stop responding.

Detect a stalled broker

Alert on these notification metrics, which MinIO AIStor reports for each server and target:

Metric	Alert when
`minio_notification_target_queue_length`	It stays above `0` for 10 minutes. Events are waiting for Kafka.
`minio_notification_target_failed_events`	It increases. The queue is full and MinIO AIStor is dropping new events.

For example, for a target with the identifier primary:

min_over_time(minio_notification_target_queue_length{target_id="kafka_primary"}[10m]) > 0
increase(minio_notification_target_failed_events{target_id="kafka_primary"}[5m]) > 0

A stalled broker can look healthy to Kafka. It keeps passing Kafka’s own heartbeats while its client listener fails, for example with TLS handshake resets or a firewall rule. To catch this, test each broker’s client listener with a real produce request, not only a TCP connection check.

Recover from a stalled broker

Stop the broker process that fails client requests. Kafka then moves its partitions to healthy brokers, and MinIO AIStor sends the queued events without further action. Repair the broker’s client listener before you return it to the cluster.

Removing the broker from the MinIO AIStor brokers list does not help. MinIO AIStor uses that list only to discover the cluster, and Kafka still sends producers to the broker for the partitions it leads. Restarting MinIO AIStor does not help either, because the queued events go to the same broker.

Size the queue for the outage you want to survive

queue_limit applies to each target on each server. Set it to the peak per-server event rate for the target, multiplied by the number of seconds of outage you want to absorb, plus headroom for bursts.

For example, at a peak of 50 events per second per server, a one-hour outage needs 180,000 entries. Set queue_limit to 200000 and confirm that queue_dir has enough free space.

When the queue is full, MinIO AIStor drops new events for that target. Events dropped this way are not replayed after Kafka recovers.

Publish Events to Kafka

On this page