Configure Metadata Service (MDS)

The Confluent Platform Metadata Service (MDS) manages a variety of metadata about your Confluent Platform installation. Specifically, the MDS:

  • Hosts the cluster registry that enables you to keep track of which clusters you have installed.
  • Serves as the system of record for cross-cluster authorization data (including RBAC, and centralized ACLs), and can be used for token-based authentication.
  • Provides a convenient way to manage audit log configurations across multiple clusters.
  • Can be used to authenticate data (note that client authentication is not supported).

You can set up the MDS internally within a Kafka cluster that serves other functions, and manage permissions in the same way that a database stores permissions for users logging into the database itself. You can also use the MDS to store user data. For the Kafka cluster hosting MDS, you must configure MDS on each Kafka broker, and you should synchronize these configurations across nodes.

You can also set up MDS on a dedicated Kafka cluster, servicing multiple worker Kafka clusters in such a way that security information is isolated away from client data. In the case of role-based access control (RBAC), the MDS offers a single, centralized configuration context that, after it is set up for a cluster, saves administrators from the complex and time-consuming task of defining and assigning roles for each resource on an individual basis. The MDS can enforce the rules for RBAC, centralized audit logs, centralized ACLs, and the cluster registry on its host Kafka cluster and across multiple secondary clusters (such as Kafka, Connect, and Schema Registry). So you can use a single Kafka cluster hosting MDS to manage and secure multiple secondary Kafka, Connect, Schema Registry, and ksqlDB clusters.

The MDS listens for commands using HTTP on the default port 8090. MDS maintains a local cache of authorization data that is persisted to an internal Kafka topic named _confluent-metadata-auth.

Running on a Kafka broker, you can optionally integrate MDS with LDAP to provide authentication and refreshable bearer tokens for impersonation. Note that impersonation is restricted to Confluent components. For details about configuring LDAP integration with RBAC, see Configure LDAP Authentication.

This topic includes the following configuration tasks:

Prerequisites

  • You must download self-managed Confluent Platform for your environment.

  • Active Directory (LDAP service) must be configured. The configurations in this topic are based on Microsoft Active Directory (AD). You must update these configurations to match your LDAP service.

    Note

    Nested LDAP groups are not supported.

  • Brokers running MDS must be configured with a separate listener for inter-broker communication. To access this listener, a user must be configured with ZooKeeper-based ACLs (not centralized ACLs) for authorization. If required, you can configure these users as super.users, but they cannot rely on access to resources using role-based or group-based access. The broker user must be configured as a super user or granted access using ACLs.

    Tip

    To avoid broker listener or inter-broker communication issues after you have migrated from ZooKeeper-based ACLs to centralized ACLs, you may want to include broker users in the list of super.users to ensure access. However, after verifying communications are working post-migration, be sure to remove the broker users from super.users.

  • Brokers will accept requests on the inter-broker listener port before the metadata for RBAC authorization has been initialized. However, requests on other ports are only accepted after the required metadata has been initialized, including any available LDAP metadata. Broker initialization only completes after all relevant metadata has been obtained and cached. When starting multiple brokers in an MDS cluster with a replication factor of 3 (default) for a metadata topic, at least three brokers must also be started simultaneously to enable initialization to complete on the brokers. Note that there is a timeout/retry limitation for this initialization, which you can specify in confluent.authorizer.init.timeout.ms. For details, refer to Configuring the Confluent Server Authorizer.

  • REST Proxy services that integrate with AD/LDAP using MDS will use the user login name as the user principal for authorization decisions. By default, this is also the principal used by brokers for users authenticating using SASL/GSSAPI (Kerberos). If your broker configuration overrides principal.builder.class or sasl.kerberos.principal.to.local.rules to create a different principal, the user principal used by brokers may be different from the principal used by other Confluent Platform components. In this case, you should configure ACLs and role bindings for your customized principal for broker resources.

Important

The user ID specified in group role bindings is case-specific, and must match the case specified in the AD record. Also note that when logging in as a super user, the login ID is also case-specific and must match the case specified for the user ID in role bindings.

Create a PEM key pair

You must create a PEM (Privacy-Enhanced Mail) key pair for use by the token service. This key pair is added to your server.properties file in the next section.

  1. Create a PEM key.

    Note

    The PEM key length depends on the encryption method you are using (AES-256 or RSA-4096), and the bits should be based on the needs/requirements of the method being used.

    This example shows how to create the 2048-bit RSA private key, and stores the keys in a folder named <path-to-tokenKeypair.pem> (replace the path in <> to reflect your setup).

    mkdir <path-to-tokenKeypair.pem> && openssl genrsa -out <path-to-tokenKeypair.pem> 2048

  2. Extract the public key.

    openssl rsa -in <path-to-tokenKeypair.pem> -outform PEM -pubout -out <path-to-public.pem>

Note

Only use OpenSSL to create the PEM key files. Do not use Keytool; it is not a valid option and will result in an error during startup.

Configure a primary Kafka cluster to host the MDS and role binding

Tip

You can store passwords and other configuration data securely using the Confluent CLI confluent secret commands. For more information refer to Secrets Management.

Note

You must complete and run the MDS configuration steps described here on all available brokers in your cluster. The configuration example here is for a cluster with a standalone broker. To view an example of a multiple broker (inter-broker) configuration, refer to Configure mTLS Authentication and RBAC for Kafka Brokers.

The following sections describe how to configure a primary Kafka cluster to host the MDS:

If you encounter issues configuring token authentication, refer to Token authentication.

Important

When all sections of the MDS configuration are complete, Start Confluent Platform.

Configure the Confluent Server Authorizer

An authorizer is a server plugin used by Apache Kafka® to authorize operations. More specifically, an authorizer controls whether or not to authorize an operation based on the principal and the resource being accessed.

The Confluent Server Authorizer supports proprietary role-based access control (RBAC) authorization for LDAP-based users and groups, as well as the setting of ACLs. Confluent Server Authorizer also supports pluggable authorization and group providers, enabling ACLs, and RBAC providers to be loaded at runtime.

Add the following configuration for Confluent Authorizer to your Kafka properties file (/etc/kafka/server.properties). Any content in brackets (<>) must be customized for your environment.

1############################# Confluent Server Authorizer Settings  #############################
2authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
3
4# Specify a list of Access Rule Providers to retain ACLs that have already been enabled and to enable RBAC
5confluent.authorizer.access.rule.providers= ZK_ACL,CONFLUENT
6
7# Specify when bootstrapping Confluent Platform and to assign a SystemAdmin
8super.users=<User:admin;User:mds>

Important

When all sections of the MDS configuration are complete, Start Confluent Platform.

The following sections describe the configuration properties used to specify the Confluent Server Authorizer settings.

authorizer.class.name

Defines the authorizer to use and serves as an entry point for Confluent proprietary functionality. In this MDS configuration, it turns “on” the Confluent Server Authorizer.

confluent.authorizer.access.rule.providers

List of access rule providers that are enabled. The Kafka Authorizer uses access rule providers to determine what rules should be used in access decisions. Supported access rule providers are:

  • CONFLUENT: Uses RBAC role bindings and Centralized ACLs stored in Kafka to generate a set of access rule objects.
  • ZK_ACL (default): Uses ACLs stored in ZooKeeper to generate a set of access rule objects.

super.users

In the MDS configuration, you should assign the super.users attribute to define a user who has full access to all resources within a Metadata Service (MDS) cluster. The primary use of super.users is to bootstrap Confluent Platform and assign a SystemAdmin. On MDS clusters, the super.user can create role bindings for all other clusters. Permissions granted by super.user apply only to the broker where the super.user attribute is specified, and not to other brokers, clusters, or Confluent Platform components. No authorization is enforced on users defined as super.users, so we strongly recommend that you specify this attribute with a limited number of users (for example, 1-2 users who are responsible for bootstrapping).

Note

Bootstrapping Confluent Platform means that when you bring up a cluster for the very first time, you are linking to or relying on another server/cluster that already includes the correct host and other configuration details needed to start up, rather than having to enter all of startup attributes each and every time you start up a new cluster. Bootstrapping saves time and resource, while providing a reliable cluster startup experience.

Configure the LDAP identity provider

The basic LDAP configuration for MDS is described below. This configuration shows the LDAP context to identify LDAP users and groups to the MDS. Be aware that nested LDAP groups are not supported.

Be prepared to provide the following information, which you will need to specify in your LDAP configuration:

  • The host (LDAP server URL, for example, LDAPSERVER.EXAMPLE.COM), port (for example, 389), and any other security mechanisms (such as TLS/SSL)
  • The full DN (distinguished name) of LDAP users
  • If you have a complex LDAP directory tree, consider providing search filters so that MDS can narrow down LDAP search results

Note

After configuring LDAP–but before configuring MDS–it is recommended that you connect to and query your LDAP server to verify your LDAP connection information. It is recommended that you use an LDAP tool to do this (for example, JXplorer).

Note

If you enable LDAP authentication for Kafka clients by adding the LDAP callback handler (not shown in this configuration):

  • Specify ldap.user.password.attribute only if your LDAP server does not support simple bind.
  • If you define this property (io.confluent.security.auth.provider.ldap.LdapAuthenticateCallbackHandler), LDAP will perform the user search and return the password back to Kafka and Kafka will perform the password check.
  • The LDAP server will return the user’s hashed password, so Kafka cannot authenticate the user unless the user’s properties file also uses the hashed password.

Add the following configuration for your identify provider (LDAP) to your Kafka properties file (/etc/kafka/server.properties). Any content in brackets (<>) must be customized for your environment.

 1############################# Identity Provider Settings (LDAP) #############################
 2# Search groups for group-based authorization.
 3ldap.group.name.attribute=<sAMAccountName>
 4ldap.group.object.class=group
 5ldap.group.member.attribute=member
 6ldap.group.member.attribute.pattern=CN=(.*),DC=rbac,DC=confluent,DC=io
 7ldap.group.search.base=CN=Users,DC=rbac,DC=confluent,DC=io
 8#Limit the scope of searches to subtrees off of base
 9ldap.user.search.scope=2
10#Enable filters to limit search to only those groups needed
11ldap.group.search.filter=(|(CN=<specific group>)(CN=<specific group>))
12
13# Kafka authenticates to the directory service with the bind user.
14ldap.java.naming.provider.url=ldap://<hostname>:389
15ldap.java.naming.security.authentication=simple
16ldap.java.naming.security.credentials=<password>
17ldap.java.naming.security.principal=<mds-user-DN>
18
19# Locate users. Make sure that these attributes and object classes match what is in your directory service.
20ldap.user.name.attribute=<sAMAccountName>
21ldap.user.object.class=user
22ldap.user.search.base=<user-search-base-DN>

Important

When all sections of the MDS configuration are complete, Start Confluent Platform.

The following sections provide details about the baseline LDAP configuration options for user and group-based authorization. For more details about LDAP configuration, see Configure LDAP Group-Based Authorization for MDS and Configure LDAP Authentication.

ldap.group.name.attribute

Contains the name of the group in a group entry obtained using an LDAP search. You can specify a regex pattern to extract the group name used in ACLs from this attribute by configuring ldap.group.name.attribute.pattern. The <sAMAccountName> is specific to Microsoft Active Directory and is the login name used to support clients and servers running various versions of Windows OS. Modify the value used if your LDAP configuration differs. The default for this configuration option is cn (common name).

ldap.group.object.class

Specifies the LDAP object class value that defines users in the directory service. Specify group to search groups for group-based authorization. Note that group has many applications, but is essentially a list of zero or more digital identities.

ldap.group.member.attribute

The name of the attribute that contains the members of the group in a group entry obtained using an LDAP search. The default is member. You can specify a regular expression (regex) pattern to extract the user principals from this attribute by configuring ldap.group.member.attribute.pattern.

ldap.group.member.attribute.pattern

A Java regular expression pattern that extracts the user principals of group members from group member entries obtained from the LDAP attribute specified using ldap.group.member.attribute. By default the full value of the attribute is used.

ldap.group.search.base

This attribute tells LDAP to limit the search base to group-based search using the values specified. The default is ou=groups.

ldap.group.search.scope

The LDAP search scope for group-based search. The value of 2 opens the search to include all the subtrees off the specified base, which is often too vast a space to search, and can result in timeouts. When specified, you also should specify ldap.group.search.filter. The default value is 1.

ldap.group.search.filter

The LDAP search filter for group-based search. Enables filters to limit search to only those groups needed. It is recommended that you list all the groups that will be used for searching. This is typically required because the LDAP trees in large organizations tend to be so large that trying to search it all results in timeouts. For instance, after you add a scope of 2 in ldap.group.search.scope to search all subtrees, you need to narrow the groups that are included in the search. You can include any number of groups in this search filter.

The following sections provide details about the baseline LDAP configuration options that Kafka uses to authenticate to the directory service with the bind user.

ldap.java.naming.provider.url

This option defines the URL to use for connections to the LDAP server. The default hostname is localhost; the default port is 389. You must specify this option for the MDS configuration.

ldap.java.naming.security.authentication

If password authentication is enabled on your LDAP server, you can configure the user principal and password so that brokers can authenticate with the LDAP server using simple authentication. If you do not want to authenticate with the LDAP server, specify none. The recommended value to get your MDS configuration up and running is simple, which is a PLAINTEXT authentication protocol and provides no security. For production instances, you should specify a more secure SASL method supported by your LDAP server (such as SASL_GSSAPI).

ldap.java.naming.security.credentials

Specifies the security credentials (password) of the principal performing the LDAP search.

ldap.java.naming.security.principal

Specifies the security principal, which is the distinguished name of the LDAP user performing the LDAP search. In this configuration, specify the MDS user using the DN (LDAP distinguished name, which is a sequence of relative distinguished names (RDN) connected by commas).

The following sections provide details for the options used to locate users. Make sure that these attributes and object classes match what is in your directory service.

ldap.user.name.attribute

This attribute identifies the user principal in a user entry obtained using an LDAP search. You can specify a regex pattern to extract the user principal from this attribute by configuring ldap.user.name.attribute.pattern. The <sAMAccountName> is specific to Active Directory and is the login name used to support clients and servers running various versions of Windows OS. Modify this configuration if your LDAP configuration differs. The default for this option is cn (common name).

ldap.user.object.class

Specifies the LDAP object class value that defines users in the directory service. Specify user to search for user-based authorization.

ldap.user.search.base

Use to specify the LDAP search base for a user-based search. The default value is ou=users.

Configure the MDS server

This section describes the options to create a key pair for MDS and configure MDS on the broker. You must update paths to the key files to match your setup.

Note

The Confluent Server supports the servlet applications REST Proxy and MDS.

  • If MDS is configured, then its listener configuration takes precedence over the REST Proxy listener configurations.
  • If MDS is disabled, then the REST Proxy listener configurations take precedence.

Add the following configuration for MDS server settings to your Kafka properties file (/etc/kafka/server.properties). Any content in brackets (<>) must be customized for your environment.

1############################# MDS Server Settings #############################
2# Bind Metadata Service HTTP service to port 8090.
3confluent.metadata.server.listeners=http://0.0.0.0:8090
4# The key to encrypt the token (when you issue you a token)
5confluent.metadata.server.token.key.path=<path-to-token-key-pair.pem>
6# Supported authentication methods
7confluent.metadata.server.authentication.method=BEARER

Important

When all sections of the MDS configuration are complete, Start Confluent Platform.

The following sections describe the configuration options used to specify the MDS server settings.

confluent.metadata.server.listeners

Use to bind the HTTP (or HTTPS) service to a port. The default value (8090) is specified here.

confluent.metadata.server.token.key.path

Location of the PEM-encoded public/private key pair to be used for signing and verifying tokens. Because the token service only supports RS256 signatures, key pairs must be generated using the RSA algorithm.

confluent.metadata.server.authentication.method

Use to specify that you support token authentication on the MDS side. Use BEARER to indicate that bearer token authentication is enabled for the configuration.

Configure the token listener

MDS can give a token in exchange for a user name and password. MDS can also accept a token from a user or client and authenticate that principal by the token (which can be used after the first time to continue to authenticate).

The tokens are used to authenticate to the Kafka-configured OAUTHBEARER listener. This section shows how to enable and configure the MDS token service.

  • Be sure to use the same public keys across components and brokers.
  • Do not use a unique key for each MDS service or component connecting to OAUTH.
  • Use the same private key across the brokers. In this way, MDS can properly decrypt the tokens when doing impersonation.
  • Be aware that after setting up MDS, the OAUTH listener does not display specific errors in relation to the MDS decryption keys (tokenKeypair.pem and public.pem) when they are different.

Add the following configuration for token listener settings to your Kafka properties file (/etc/kafka/server.properties). Any content in brackets (<>) must be customized for your environment.

 1############################# Token Listener Settings #############################
 2# Add named listener TOKEN to existing listeners and advertised.listeners
 3advertised.listeners=<advertised.listeners>,TOKEN://localhost:9092
 4listeners=<listeners>,TOKEN://:9092
 5# Set SASL callback handler for handling tokens on login. This is essentially a noop if not used for inter-broker communication.
 6listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler
 7# Set SASL callback handler for verifying authentication token signatures
 8listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler
 9# Configure the public key used to verify RBAC Metadata Service signatures
10# Note: username, password and metadataServerUrls must be set if used for inter-broker communication
11listener.name.token.oauthbearer.sasl.jaas.config= \
12   org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
13   publicKeyPath="<path-to-public.pem>";
14# Add protocol mapping for newly-added named listener TOKEN
15listener.security.protocol.map=<listener.map>,TOKEN:SASL_PLAINTEXT
16listener.name.token.sasl.enabled.mechanisms=OAUTHBEARER

Important

When all sections of the MDS configuration are complete, Start Confluent Platform.

The following sections describe the configuration options used to specify the token listener settings.

advertised.listeners

Comma-separated list of URIs and listener names for other brokers and clients to use. In IaaS environments, this may need to be different from the interface to which the broker binds. If this is not set, the value for listeners will be used.

listeners

Comma-separated list of listeners that listen for API requests over either HTTP or HTTPS. Each listener must include the hostname and the port.

listener.name.rbac.oauthbearer.sasl.login.callback.handler.class

Use to configure and enable Kafka API calls (for example, produce, consume) so they can understand and authenticate MDS JSON web tokens and passed usernames-passwords. For details, refer to Configuring Token Authentication.

listener.name.rbac.oauthbearer.sasl.server.callback.handler.class

Use to configure and enable Kafka API calls (for example, produce, consume) such that they will be able to understand and authenticate MDS JSON web tokens and passed usernames-passwords. For details, refer to Configuring Token Authentication.

listener.name.rbac.oauthbearer.sasl.jaas.config

Used by listener.name.rbac.oauthbearer.sasl.login.callback.handler.class to verify incoming JSON Web Tokens (JWT).

listener.security.protocol.map

This is a Kafka broker configuration option that defines key/value pairs for the security protocol to use, per listener name. Map between listener names and security protocols. This must be defined for the same security protocol to be usable in more than one port or IP. For example, internal and external traffic can be separated even if TLS/SSL is required for both. More precisely, the user could define listeners with names INTERNAL and EXTERNAL and this property as: INTERNAL:SSL,EXTERNAL:SSL. As shown, key and value are separated by a colon and map entries are separated by commas. Each listener name should only appear once in the map. Different security (TLS/SSL and SASL) settings can be configured for each listener by adding a normalized prefix (the listener name is lowercase) to the configuration name.

listener.name.rbac.sasl.enabled.mechanisms

A comma-separated list of SASL mechanisms enabled on the RBAC listener.

Important

Do not use token services or the OAUTHBEARER SASL mechanism (listener.name.rbac.sasl.enabled.mechanisms=OAUTHBEARER) for external client communications. With RBAC enabled, token services are intended for internal communication between Confluent Platform components only (for example, it is valid for a Schema Registry licensed client), and not for long-running service principals or client authentication. The OAUTHBEARER setting is for internal use and subject to change, and does not implement a full-featured OAuth protocol. Therefore, use one of the supported authentication methods like SASL or mTLS (mutual TLS) for long-lived or client use cases. For details, refer to Authentication Methods Overview.

Important

When all sections of the MDS configuration are complete, Start Confluent Platform.

Full primary Kafka cluster MDS configuration

This example shows the full configuration for the primary Kafka cluster that is hosting MDS and role binding:

 1############################# Confluent Server Authorizer Settings  #############################
 2authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
 3
 4# Specify a list of Access Rule Providers to retain ACLs that have already been enabled and to enable RBAC
 5confluent.authorizer.access.rule.providers= ZK_ACL,CONFLUENT
 6
 7# Specify when bootstrapping Confluent Platform and to assign a SystemAdmin
 8super.users=<User:admin;User:mds>
 9
10############################# Identity Provider Settings (LDAP) #############################
11# Search groups for group-based authorization.
12ldap.group.name.attribute=<sAMAccountName>
13ldap.group.object.class=group
14ldap.group.member.attribute=member
15ldap.group.member.attribute.pattern=CN=(.*),DC=rbac,DC=confluent,DC=io
16ldap.group.search.base=CN=Users,DC=rbac,DC=confluent,DC=io
17#Limit the scope of searches to subtrees off of base
18ldap.user.search.scope=2
19#Enable filters to limit search to only those groups needed
20ldap.group.search.filter=(|(CN=<specific group>)(CN=<specific group>))
21
22# Kafka authenticates to the directory service with the bind user.
23ldap.java.naming.provider.url=ldap://<hostname>:389
24ldap.java.naming.security.authentication=simple
25ldap.java.naming.security.credentials=<password>
26ldap.java.naming.security.principal=<mds-user-DN>
27
28# Locate users. Make sure that these attributes and object classes match what is in your directory service.
29ldap.user.name.attribute=<sAMAccountName>
30ldap.user.object.class=user
31ldap.user.search.base=<user-search-base-DN
32
33############################# MDS Server Settings #############################
34# Bind Metadata Service HTTP service to port 8090.
35confluent.metadata.server.listeners=http://0.0.0.0:8090
36# The key to encrypt the token (when you issue you a token)
37confluent.metadata.server.token.key.path=<path-to-token-key-pair.pem>
38# Supported authentication methods
39confluent.metadata.server.authentication.method=BEARER
40
41############################# Token Listener Settings #############################
42# Add named listener TOKEN to existing listeners and advertised.listeners
43advertised.listeners=<advertised.listeners>,TOKEN://localhost:9092
44listeners=<listeners>,TOKEN://:9092
45# Set SASL callback handler for handling tokens on login. This is essentially a noop if not used for inter-broker communication.
46listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler
47# Set SASL callback handler for verifying authentication token signatures
48listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler
49# Configure the public key used to verify RBAC Metadata Service signatures
50# Note: username, password and metadataServerUrls must be set if used for inter-broker communication
51listener.name.token.oauthbearer.sasl.jaas.config= \
52   org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
53   publicKeyPath="<path-to-public.pem>";
54# Add protocol mapping for newly-added named listener TOKEN
55listener.security.protocol.map=<listener.map>,TOKEN:SASL_PLAINTEXT
56listener.name.token.sasl.enabled.mechanisms=OAUTHBEARER

Troubleshooting MDS configuration issues

The following sections provide guidance to help your troubleshoot issues you may encounter in your MDS configuration.

Token authentication

Token login and authentication to MDS using OAUTHBEARER can fail without obvious known exceptions or errors appearing in either server.log or metadata-service.log. Such failures may be due to either the public and private key files being corrupt, or because the token was generated using different public and private key files.

To troubleshoot token authentication:

  1. Delete both tokenKeypair.pem and public.pem from the folders configured in server.properties and regenerate them. The newly-generated key files will be placed in:

    confluent.metadata.server.token.key.path=<path-to-token-key-pair.pem>

  2. After regenerating tokenKeypair.pem and public.pem, restart the broker server where MDS is running.

  3. On the client machine, delete the local CLI cache (~/.confluent/config.json), which cached the token for the super user after the super user logged in.

  4. Log in again using the CLI.

MDS REST client configurations

If a component (such as Schema Registry, ksqlDB, Confluent Control Center, or Connect) client configured to communicate with MDS includes an incorrect username or password, it can result in an endless loop of attempts to authenticate, which can inadvertently produce a continuous loop of exceptions in your REST client exception log and impact performance. For example:

[2021-01-25 05:11:58,330] ERROR [pool-17-thread-1] Error while refreshing active metadata server urls, retrying (io.confluent.security.auth.client.rest.RestClient)
io.confluent.security.auth.client.rest.exceptions.RestClientException: Unauthorized; error code: 401
   at io.confluent.security.auth.client.rest.RestClient$HTTPRequestSender.lambda$submit$0(RestClient.java:353)
   at java.util.concurrent.FutureTask.run(FutureTask.java:266)
   at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
   at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
   at java.lang.Thread.run(Thread.java:748)

To control the number of authentication retry attempts, include the following options in your REST client MDS configuration:

  • confluent.metadata.server.urls.max.retries
  • confluent.metadata.server.urls.fail.on.401

For details about these configuration options, refer to REST client configurations.

Configure a secondary Kafka cluster managed by the MDS of the primary Kafka cluster

The following sections describe how to configure a secondary Kafka cluster managed by the MDS of the primary Kafka cluster.

Important

When all sections of the MDS configuration are complete, Start Confluent Platform.

Configure the Confluent Server Authorizer

See Configure the Confluent Server Authorizer for details about the Confluent Server Authorizer and the settings recommended here.

Add the following MDS configuration to your Kafka properties file (/etc/kafka/server.properties). Any content in brackets (<>) must be customized for your environment.

1############################# Confluent Server Authorizer Settings  #############################
2authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
3confluent.authorizer.access.rule.providers= ZK_ACL,CONFLUENT

Important

When all sections of the MDS configuration are complete, Start Confluent Platform.

Configure the MDS

This section configures Kafka so that it can talk to the MDS cluster and “consume” the role bindings. In the configuration example below, SASL_PLAINTEXT/PLAIN is used, but you should use whatever security mechanism is required by the Kafka broker running MDS. Also, you should specify the port exposed by the other broker. For more information, see Metadata Service Configuration Options.

1############################# MDS Settings #############################
2confluent.metadata.bootstrap.servers=<kafka-with-mds-host-1>:<port>,<kafka-with-mds-host-2>:<port>,...
3confluent.metadata.security.protocol=SASL_PLAINTEXT
4confluent.metadata.sasl.mechanism=PLAIN
5confluent.metadata.sasl.jaas.config= \
6   org.apache.kafka.common.security.plain.PlainLoginModule required \
7   username="<broker-username>" \
8   password="<broker-password>";

Important

When all sections of the MDS configuration are complete, Start Confluent Platform.

The following sections describe the configuration options used to specify the MDS settings.

confluent.metadata.bootstrap.servers

The Kafka host and port of the cluster host MDS. Used to determine how a broker connects to the MDS. This option is required for managed brokers, and optional for MDS brokers, which default to the inter-broker values. An example value looks like ec2-22-222-22-222.compute-1.amazonaws.com:9092.

confluent.metadata.security.protocol

Defines the security options to be used when connecting to an external MDS, and provides managed brokers the ability to consume RBAC data.

confluent.metadata.sasl.mechanism

Defines the security options to be used when connecting to an external MDS, and provides managed brokers the ability to consume RBAC data.

confluent.metadata.sasl.jaas.config

Defines the JAAS configuration for managed clusters to connect to and consume the role binding data so that they can locally enforce RBAC on direct Kafka API calls to themselves.

Configure the token listener

This section of the MDS configuration enables the listener with the OAUTHBEARER SASL mechanism, which is used for impersonation. For more information, see Recommended Kafka Broker configuration. For details about token listener settings in an MDS configuration, see Configure the token listener.

Any differences between the token listener settings used here and those described above in Configure the token listener are detailed below.

1############################# Token-based Listener Settings #############################
2listeners=<listeners>,TOKEN://:9092
3advertised.listeners=<advertised.listeners>,TOKEN://<hostname>:9092
4listener.security.protocol.map=<advertised.listeners>,TOKEN:SASL_PLAINTEXT
5listener.name.token.oauthbearer.sasl.jaas.config= \
6   org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
7   publicKeyPath="<path-to-public.pem>";
8listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler
9listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler

listener.name.token.oauthbearer.sasl.jaas.config

The oauthbearer.sasl.jaas.config of the defined LISTENER_NAME. Here, TOKEN is the name of the listener. In other words, listener.name.token.oauthbearer.sasl.jaas.config is an instance of listener.name.<LISTENER_NAME>.oauthbearer.sasl.jaas.config.

Important

When all sections of the MDS configuration are complete, Start Confluent Platform.

Full secondary Kafka cluster MDS configuration

This example shows the full configuration for the secondary Kafka cluster that is managed by the MDS of the primary Kafka cluster:

 1############################# Confluent Server Authorizer Settings  #############################
 2authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
 3confluent.authorizer.access.rule.providers= ZK_ACL,CONFLUENT
 4
 5############################# MDS Settings #############################
 6confluent.metadata.bootstrap.servers=<kafka-with-mds-host-1>:<port>,<kafka-with-mds-host-2>:<port>,...
 7confluent.metadata.security.protocol=SASL_PLAINTEXT
 8confluent.metadata.sasl.mechanism=PLAIN
 9confluent.metadata.sasl.jaas.config= \
10   org.apache.kafka.common.security.plain.PlainLoginModule required \
11   username="<broker-username>" \
12   password="<broker-password>";
13
14############################# Token-based Listener Settings #############################
15listeners=<listeners>,TOKEN://:9092
16advertised.listeners=<advertised.listeners>,TOKEN://<hostname>:9092
17listener.security.protocol.map=<advertised.listeners>,TOKEN:SASL_PLAINTEXT
18listener.name.token.oauthbearer.sasl.jaas.config= \
19   org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
20   publicKeyPath="<path-to-public.pem>";
21listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler
22listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler