Connect Gateway to Kafka

Gateway depends on a ‘backing’ Kafka cluster for its operation. Configuring the Gateway connection to the backing Kafka cluster closely resembles configuring a standard Kafka client’s connection to a cluster. Set the client to Gateway configuration variables first. Gateway has to know which security mode you run, because that determines whether it reuses each client’s credentials to reach Kafka or connects with its own service account. The configuration is done via environment variables, as it is for the other aspects of a Gateway configuration. All environment variables that start with KAFKA_ are mapped to configuration properties for connecting Gateway to the Kafka cluster. As Gateway is based on the Java-based Kafka-clients, it supports all configuration properties that Java-clients do. Kafka configuration properties are mapped to Gateway environment variables as follows:

Add a KAFKA_ prefix
Replace each dot, . , with an underscore, _
Convert to uppercase

For example, bootstrap.servers is set by the KAFKA_BOOTSTRAP_SERVERS environment variable.

Supported protocols

You can use all the Kafka security protocols to authenticate Gateway to the Kafka cluster; PLAINTEXT, SASL_PLAINTEXT, SASL_SSL and SSL. These can be used with all SASL mechanisms supported by Apache Kafka: PLAIN, SCRAM-SHA, OAuthBearer, Kerberos etc. In addition, we support IAM authentication for AWS MSK clusters. In the following examples, we provide blocks of environment variables which can be provided to Gateway, e.g. in a docker-compose file, or a helm deployment. Information which should be customized is enclosed by < and >.

PLAINTEXT

Kafka cluster without authentication or encryption in transit, PLAINTEXT. In this case you just need the bootstrap servers:

KAFKA_BOOTSTRAP_SERVERS: <your.kafka.broker-1:9092>,<your.kafka.broker-2:9092>

SSL

KAFKA_BOOTSTRAP_SERVERS: <your.kafka.broker-1:9092>,<your.kafka.broker-2:9092>
KAFKA_SECURITY_PROTOCOL: SSL
KAFKA_SSL_TRUSTSTORE_TYPE: <JKS>
KAFKA_SSL_TRUSTSTORE_LOCATION: </path/to/truststore.jks>
KAFKA_SSL_TRUSTSTORE_PASSWORD: <truststore-password>
KAFKA_SSL_KEYSTORE_TYPE: <JKS>
KAFKA_SSL_KEYSTORE_LOCATION: </path/to/keystore.jks>
KAFKA_SSL_KEYSTORE_PASSWORD: <keystore-password>
KAFKA_SSL_KEY_PASSWORD: <key-password>

mTLS

Kafka cluster with mTLS client authentication.

KAFKA_BOOTSTRAP_SERVERS: <your.kafka.broker-1:9092>,<your.kafka.broker-2:9092>, <your.kafka.broker-3:9092>
KAFKA_SECURITY_PROTOCOL: SSL
KAFKA_SSL_TRUSTSTORE_LOCATION: /security/truststore.jks
KAFKA_SSL_TRUSTSTORE_PASSWORD: conduktor
KAFKA_SSL_KEYSTORE_LOCATION: /security/kafka.gw.keystore.jks
KAFKA_SSL_KEYSTORE_PASSWORD: conduktor
KAFKA_SSL_KEY_PASSWORD: conduktor

SASL_PLAINTEXT

Kafka cluster with SASL_PLAINTEXT security protocol but no encryption in transit, supporting the following SASL_MECHANISMs.

SASL PLAIN

KAFKA_BOOTSTRAP_SERVERS: <your.kafka.broker-1:9092>,<your.kafka.broker-2:9092>
KAFKA_SECURITY_PROTOCOL: SASL_PLAINTEXT
KAFKA_SASL_MECHANISM: PLAIN
KAFKA_SASL_JAAS_CONFIG: org.apache.kafka.common.security.plain.PlainLoginModule required username="<gw-sa-username>" password="<gw-sa-password>";

SASL SCRAM

KAFKA_BOOTSTRAP_SERVERS: <your.kafka.broker-1:9092>,<your.kafka.broker-2:9092>
KAFKA_SECURITY_PROTOCOL: SASL_PLAINTEXT
KAFKA_SASL_MECHANISM: SCRAM-SHA-256 # or SCRAM-SHA-512
KAFKA_SASL_JAAS_CONFIG: org.apache.kafka.common.security.scram.ScramLoginModule required username="<gw-sa-username>" password="<gw-sa-password>";

OAuthBearer (client secret)

KAFKA_BOOTSTRAP_SERVERS: <your.kafka.broker-1:9092>,<your.kafka.broker-2:9092>
KAFKA_SECURITY_PROTOCOL: SASL_PLAINTEXT
KAFKA_SASL_MECHANISM: OAUTHBEARER
KAFKA_SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL: "<https://myidp.example.com/oauth2/default/v1/token>"
KAFKA_SASL_LOGIN_CALLBACK_HANDLER_CLASS: org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
KAFKA_SASL_JAAS_CONFIG: org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="<>" clientSecret="<>" scope=".default";

SASL_SSL

Kafka cluster that uses SASL for authentication and TLS (formerly SSL) for encryption in transit.

PLAIN

Kafka cluster with SASL_SSL and PLAIN SASL mechanism.

Confluent Cloud with API key/secret

This example can be seen as a special case of the above.

KAFKA_BOOTSTRAP_SERVERS: <your.cluster.confluent.cloud:9092>
KAFKA_SECURITY_PROTOCOL: SASL_SSL
KAFKA_SASL_MECHANISM: PLAIN
KAFKA_SASL_JAAS_CONFIG: org.apache.kafka.common.security.plain.PlainLoginModule required username="<API_KEY>" password="<API_SECRET>";

As Confluent Cloud uses certificates signed by a well-known CA, you normally do not need to specify a trust-store.

SASL SCRAM

KAFKA_BOOTSTRAP_SERVERS: <your.kafka.broker-1:9092>,<your.kafka.broker-2:9092>
KAFKA_SECURITY_PROTOCOL: SASL_SSL
KAFKA_SASL_MECHANISM: SCRAM-SHA-256 # or SCRAM-SHA-512
KAFKA_SASL_JAAS_CONFIG: org.apache.kafka.common.security.scram.ScramLoginModule required username="<gw-sa-username>" password="<gw-sa-password>";
KAFKA_SSL_TRUSTSTORE_TYPE: <JKS>
KAFKA_SSL_TRUSTSTORE_LOCATION: </path/to/truststore.jks>
KAFKA_SSL_TRUSTSTORE_PASSWORD: <truststore-password>

SASL GSSAPI (Kerberos)

KAFKA_BOOTSTRAP_SERVERS: <your.kafka.broker-1:9092>,<your.kafka.broker-2:9092>
KAFKA_SECURITY_PROTOCOL: SASL_SSL
KAFKA_SASL_MECHANISM: GSSAPI
KAFKA_SASL_JAAS_CONFIG: com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="<>>" principal="<>";
KAFKA_SASL_KERBEROS_SERVICE_NAME: <kafka>
KAFKA_SSL_TRUSTSTORE_TYPE: <JKS>
KAFKA_SSL_TRUSTSTORE_LOCATION: </path/to/truststore.jks>
KAFKA_SSL_TRUSTSTORE_PASSWORD: <truststore-password>

OAuthBearer (client secret)

KAFKA_BOOTSTRAP_SERVERS: <your.kafka.broker-1:9092>,<your.kafka.broker-2:9092>
KAFKA_SECURITY_PROTOCOL: SASL_SSL
KAFKA_SASL_MECHANISM: OAUTHBEARER
KAFKA_SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL: "<https://myidp.example.com/oauth2/default/v1/token>"
KAFKA_SASL_LOGIN_CALLBACK_HANDLER_CLASS: org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
KAFKA_SASL_JAAS_CONFIG: org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="<>" clientSecret="<>" scope=".default";

OAuthBearer (client assertion)

Rather than a shared client secret, the Gateway can authenticate to the Kafka cluster with a signed JWT assertion (KIP-1258).

KAFKA_BOOTSTRAP_SERVERS: <your.kafka.broker-1:9092>,<your.kafka.broker-2:9092>
KAFKA_SECURITY_PROTOCOL: SASL_SSL
KAFKA_SASL_MECHANISM: OAUTHBEARER
KAFKA_SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL: "<https://myidp.example.com/oauth2/token>"
KAFKA_SASL_LOGIN_CALLBACK_HANDLER_CLASS: org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
KAFKA_SASL_JAAS_CONFIG: org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="<your-client-id>";
KAFKA_SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE: </path/to/private-key.pem>
KAFKA_SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS: <your-client-id>
KAFKA_SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB: <your-client-id>

The private key file has to be added to the allowed.files list via JAVA_OPTS:

JAVA_OPTS: "-Dorg.apache.kafka.sasl.oauthbearer.allowed.files=</path/to/private-key.pem>"

AWS MSK cluster with IAM

KAFKA_BOOTSTRAP_SERVERS: <b-3-public.****.kafka.eu-west-1.amazonaws.com:9198>
KAFKA_SECURITY_PROTOCOL: SASL_SSL
KAFKA_SASL_MECHANISM: AWS_MSK_IAM
KAFKA_SASL_JAAS_CONFIG: software.amazon.msk.auth.iam.IAMLoginModule required;
KAFKA_SASL_CLIENT_CALLBACK_HANDLER_CLASS: io.conduktor.aws.IAMClientCallbackHandler
KAFKA_AWS_ACCESS_KEY_ID: <access-key-id>
KAFKA_AWS_SECRET_ACCESS_KEY: <secret-access-key>

Service account and ACL requirements

Gateway connects to the backing Kafka with its own service account. What else that account needs depends on the security mode you run:

In Kafka-managed mode, clients present their own credentials and Gateway reuses them to reach Kafka, so each client inherits the ACLs of its own service account on the backing cluster
In Gateway-managed mode, Gateway reaches Kafka with its own service account for every client, so that account also needs every permission your Gateway users rely on

In both modes, Gateway’s own service account needs these ACLs to operate correctly:

Read, Write and DescribeConfigs on Gateway’s internal topics, which default to the _conduktor_${GATEWAY_CLUSTER_ID}_ prefix
Create on those topics, or on the cluster, unless you pre-create them all with cleanup.policy=compact on the state topics
Read on the conduktor_${GATEWAY_CLUSTER_ID} consumer group, or on the group set in GATEWAY_GROUP_ID
Describe on the physical topics behind alias topics and topic views
Write and Create on the topics Gateway writes its own records to: the audit log, the observability Interceptor (chargeback) topic, and the violation and dead letter topics of your data validation policies

With topic concentration, it also needs:

Describe and DescribeConfigs on the physical topics, plus Create and AlterConfigs for auto-managed rules
DescribeConfigs on the cluster, if clients create concentrated topics without a partition count
Read on the physical topics behind rules that still set the deprecated offsetCorrectness, and a prefixed Read ACL on consumer group conduktor_${GATEWAY_CLUSTER_ID}-logical-offsets-, or ${GATEWAY_GROUP_ID}-logical-offsets- when you set that variable

Conduktor in production

Deploy Console

Deploy Schema Registry Proxy

Deploy Gateway

Initial setup

Supported protocols

PLAINTEXT

SSL

mTLS

SASL_PLAINTEXT

SASL PLAIN

SASL SCRAM

OAuthBearer (client secret)

SASL_SSL

PLAIN

Confluent Cloud with API key/secret

SASL SCRAM

SASL GSSAPI (Kerberos)

OAuthBearer (client secret)

OAuthBearer (client assertion)

AWS MSK cluster with IAM

Service account and ACL requirements

​Supported protocols

​PLAINTEXT

​SSL

​mTLS

​SASL_PLAINTEXT

​SASL PLAIN

​SASL SCRAM

​OAuthBearer (client secret)

​SASL_SSL

​PLAIN

​Confluent Cloud with API key/secret

​SASL SCRAM

​SASL GSSAPI (Kerberos)

​OAuthBearer (client secret)

​OAuthBearer (client assertion)

​AWS MSK cluster with IAM

​Service account and ACL requirements

​Related resources

Supported protocols

PLAINTEXT

SSL

mTLS

SASL_PLAINTEXT

SASL PLAIN

SASL SCRAM

OAuthBearer (client secret)

SASL_SSL

PLAIN

Confluent Cloud with API key/secret

SASL SCRAM

SASL GSSAPI (Kerberos)

OAuthBearer (client secret)

OAuthBearer (client assertion)

AWS MSK cluster with IAM

Service account and ACL requirements

Related resources