Kafka Producer

The Kafka Producer output plugin lets you ingest your records into an Apache Kafka service. This plugin uses the official librdkafka C library.

In Fluent Bit 4.0.4 and later, the Kafka input plugin supports authentication with AWS MSK IAM, enabling integration with Amazon MSK (Managed Streaming for Apache Kafka) clusters that require IAM-based access.

Configuration parameters

This plugin supports the following parameters:

Setting rdkafka.log.connection.close to false and rdkafka.request.required.acks to 1 are examples of recommended settings of librdfkafka properties.

Get started

To insert records into Apache Kafka, you can run the plugin from the command line or through the configuration file.

Command line

The Kafka plugin can read parameters through the -p argument (property):

Configuration file

In your main configuration file append the following:

Avro support

Fluent Bit can encode records as Avro messages for the out_kafka plugin, but this support isn't included in any official release packages or container images. To use it, you must build Fluent Bit from source.

Avro support is optional and must be activated at build time by using a build definition with cmake: -DFLB_AVRO_ENCODER=On such as in the following example which activates:

• out_kafka with Avro encoding

• Fluent Bit Prometheus

• Metrics using an embedded HTTP endpoint

• Debugging support

• Builds the test suites

Kafka configuration file with Avro encoding

In this example, the Fluent Bit configuration tails Kubernetes logs, updates the log lines with Kubernetes metadata using the Kubernetes filter. It then sends the updated log lines to a Kafka broker encoded with a specific Avro schema.

Protobuf support

Protobuf support is available in Fluent Bit version 5.1.3 and greater. Fluent Bit can encode records as Protobuf messages for the out_kafka plugin, but this support isn't included in any official release packages or container images. To use it, you must build Fluent Bit from source.

Protobuf support is optional and must be activated at build time with -DFLB_PROTOBUF_ENCODER=On. The build requires Protobuf 3.12 or greater, including the libprotoc development libraries. In a build without it, setting format to protobuf fails at startup with format protobuf requires FLB_PROTOBUF_ENCODER=On.

Unlike Avro, Protobuf has no inline schema option. The schema always comes from a Confluent Schema Registry, so schema_registry_url is required. Without it, the plugin fails at startup with format protobuf requires schema_registry_url. See Resolve schemas from a registry.

The schema resolved from the registry becomes the root .proto file. If that schema declares references, Fluent Bit fetches each referenced subject from the registry and registers it as an imported file, resolving references recursively. A schema graph is limited to 64 files and 4 MiB in total.

Use protobuf_message to select which message encodes your records:

• If the root schema declares exactly one message, you can leave protobuf_message unset.

• Otherwise, set it to the fully qualified message name, such as com.example.logs.LogRecord.

The message must be declared in the root schema itself, not in one of its imported files. If it isn't found, the plugin fails at startup with cannot compile registered Protobuf schema: protobuf_message must select a message in the registered root schema.

Fluent Bit converts each record to JSON and then encodes it with the selected message, so record field names must match the fields declared in the schema. Encoded messages use the Confluent wire format selected by schema_registry_framing, which prefixes the payload with the schema ID and the message index path.

The following example encodes records with a Protobuf schema resolved from a registry subject:

Resolve schemas from a registry

Resolving schemas from a Confluent Schema Registry is available in Fluent Bit version 5.1 and greater. It requires either the Avro encoder build option (-DFLB_AVRO_ENCODER=On) or, in version 5.1.3 and greater, the Protobuf encoder build option (-DFLB_PROTOBUF_ENCODER=On). Neither option is included in official release builds. See Avro support and Protobuf support.

With format set to avro, the registry is an alternative to setting schema_str and schema_id in your configuration. With format set to protobuf, the registry is the only source of the schema. Set schema_registry_url to enable registry resolution. For Avro, if schema_str and schema_id are both set, Fluent Bit uses them and never contacts the registry.

Fluent Bit resolves the schema in one of two ways:

• If schema_registry_subject is set, it requests /subjects/{subject}/versions/{version}, where the version comes from schema_registry_version and defaults to latest.

• Otherwise, it requests /schemas/ids/{id} using the value of schema_id.

The schema is fetched once and reused for the lifetime of the plugin instance. If the registry can't be reached or returns an error, the chunk is retried.

To authenticate, set either schema_registry_http_user and schema_registry_http_passwd for HTTP basic authentication, or schema_registry_bearer_token for bearer token authentication.

For high availability, set schema_registry_url to a comma-separated list of registry URLs. During the initial schema fetch, Fluent Bit tries the next endpoint in the list when a request fails. Once the schema is successfully resolved, it's cached for the lifetime of the plugin instance, and the registry isn't contacted again.

Except for schema_registry_framing, each of these settings also accepts a dotted spelling that matches Confluent client configuration, such as schema.registry.url for schema_registry_url. The two spellings are interchangeable.

The following example resolves the latest version of the fluent-bit-logs-value subject from either of two registry endpoints:

Kafka configuration file with raw format

This example Fluent Bit configuration file creates example records with the payloadkey and msgkey keys. The msgkey value is used as the Kafka message key, and the payloadkey value as the payload.

AWS MSK IAM authentication

Fluent Bit 4.0.4 and later supports authentication to Amazon MSK (Managed Streaming for Apache Kafka) clusters using AWS IAM for the Kafka output plugin. This lets you securely send data to MSK brokers with AWS credentials, leveraging IAM roles and policies for access control.

Prerequisites

If you are compiling Fluent Bit from source, ensure the following requirements are met to enable AWS MSK IAM support:

• Build Requirements

• Runtime Requirements:

  • Network Access: Fluent Bit must be able to reach your MSK broker endpoints (AWS VPC setup).

  • AWS Credentials: Provide credentials using any supported AWS method:

    • IAM roles (recommended for EC2, ECS, or EKS)

    • Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)

    • AWS credentials file (~/.aws/credentials)

    • Instance metadata service (IMDS)

  These credentials are discovered by default when aws_msk_iam flag is enabled.

• IAM Permissions: The credentials must allow access to the target MSK cluster.

AWS MSK IAM configuration parameters

See aws_msk_iam and aws_msk_iam_cluster_arn in the configuration parameters table.

Configuration example

AWS IAM policy

IAM policies and permissions can be complex and can vary depending on your organization's security requirements. If you are unsure about the correct permissions or best practices, consult with your AWS administrator or an AWS expert who is familiar with MSK and IAM security.

The AWS credentials used by Fluent Bit must have permission to connect to your MSK cluster. Here is a minimal example policy: