Kroxylicious

The Virtual Cluster is the downstream representation of a Kafka Cluster. At the conceptual level, a Kafka Client connects to a Virtual Cluster. Kroxylicious proxies all communications made to the Virtual Cluster through to a (physical) Kafka Cluster, passing it through the Filter Chain.

So far, this explanation has elided the detail of Kafka Brokers. Let’s talk about that now.

The Virtual Cluster automatically exposes a bootstrap endpoint for the Virtual Cluster. This is what the Kafka Client must specify as the bootstrap.servers property in the client configuration.

In addition to the bootstrap endpoint, Kroxylicious automatically exposes broker endpoints. There is one broker endpoint for each broker of the physical cluster. When the Client connects to a broker endpoint, Kroxylicious proxies all communications to the corresponding broker of the (physical) Kafka Cluster.

Kroxylicious automatically intercepts all the Kafka RPC responses that contain a broker address. It rewrites the address so that it refers to the corresponding broker endpoint of the Virtual Cluster. This means when the Kafka Client goes to connect to, say broker 0, it does so through the Virtual Cluster.

The Target Cluster is the definition of physical Kafka Cluster within the Kroxylicious itself.

A Virtual Cluster has exactly one Target Cluster.

There can be a one-to-one relationship between Virtual Clusters and Target Clusters. The other possibility is many-to-one, where many Virtual Clusters point to the same Target Cluster. The many-to-one pattern is exploited by filters such as the Multi-tenancy filter.

A one-to-many pattern, where one Virtual Cluster points to many Target Clusters (providing amalgamation), is not a supported use-case.

A Filter Chain consists of an ordered list of pluggable protocol filters.

A protocol filter implements some logic for intercepting, inspecting and/or manipulating Kafka protocol messages. Kafka protocol requests (such as Produce requests) pass sequentially through each of the protocol filters in the chain, beginning with the 1st filter in the chain and then following with the subsequent filters, before being forwarded to the broker.

When the broker returns a response (such as a Produce response) the protocol filters in the chain are invoked in the reverse order (that is, beginning with the nth filter in the chain, then the n-1th and so on) with each having the opportunity to inspect and/or manipulating the response. Eventually a response is returned to the client.

The description above describes only the basic capabilities of the protocol filter. Richer features of filters are described later.

As mentioned above, Kroxylicious takes the responsibility to rewrite the Kafka RPC responses that carry broker address information so that they reflect the broker addresses exposed by the Virtual Cluster. These are the Metadata, DescribeCluster and FindCoordinator responses. This processing is entirely transparent to the work of the protocol filters. Filter authors are free to write their own filters that intercept these responses too.

An important principal for the protocol filter API is that filters should compose nicely. That means that filters generally don’t know what other filters might be present in the chain, and what they might be doing to messages. When a filter forwards a request or response it doesn’t know whether the message is being sent to the next filter in the chain, or straight back to the client.

Such composition is important because it means a proxy user can configure multiple filters (possibly written by several filter authors) and expect to get the combined effect of all of them.

It’s never quite that simple, of course. In practice they will often need to understand what each filter does in some detail in order to be able to operate their proxy properly, for example by understanding whatever metrics each filter is emitting.

The filter encrypts records from produce requests as follows:

The filter uses a DEK reuse strategy. Encrypted records are sent to the same topic using the same DEK until a time-out or an encryption limit is reached.

The filter decrypts records from fetch responses as follows:

The filter uses an LRU (least recently used) strategy for caching decrypted DEKs. Decrypted DEKs are kept in memory to reduce interactions with the KMS.

To support the filter, the KMS provides the following:

KEKs stay within the KMS. The KMS generates a DEK (which is securely generated random data) for a given KEK, then returns the DEK and an encrypted DEK. The encrypted DEK has the same data but encrypted with the KEK. The KMS doesn’t store encrypted DEKs; they are stored as part of the cipher record in the broker.

Key rotation involves periodically replacing cryptographic keys with new ones and is considered a best practice in cryptography.

The filter allows the rotation of Key Encryption Keys (KEKs) within the Key Management System (KMS). When a KEK is rotated, the new key material is eventually used for newly produced records. Existing records, encrypted with older KEK versions, remain decryptable as long as the previous KEK versions are still available in the KMS.

When the KEK is rotated in the external KMS, it will take up to 1 hour (by default) before all^[1] records produced by the filter contain a DEK encrypted with the new key material. This is because existing encrypted DEKs are used for a configurable amount of time after creation, the Filter caches the encrypted DEK, one hour after creation they are eligible to be refreshed.

If you need to rotate key material immediately, execute a rolling restart of your cluster of Kroxylicious instances.

The record encryption filter encrypts only the values of records, leaving record keys, headers, and timestamps untouched. Null record values, which might represent deletions in compacted topics, are transmitted to the broker unencrypted. This approach ensures that compacted topics function correctly.

You may configure the system so that some topics are encrypted and others are not. This supports scenarios where topics with confidential information are encrypted and Kafka topics with non-sensitive information can be left unencrypted.

The filter integrates with the HashiCorp Vault Transit Engine. Vault does not enable the Transit Engine by default. It must be enabled before it can be used with the filter.

For HashiCorp Vault, the KMS configuration looks like this. Use the Vault Token and Vault Transit Engine URL values that you gathered above.

For TLS trust and TLS client authentication configuration, the filter accepts the same TLS parameters as Upstream TLS except the PEM store type is currently not supported.

As the administrator, use either the HashiCorp UI or CLI to create AES-256 symmetric keys following your key naming convention. The key type must be aes256-gcm96, which is Vault’s default key type.

If using the Vault CLI, the command will look like:

The filter references KEKs within AWS via an AWS key alias.

Establish a naming convention for key aliases to keep the filter’s keys separate from those used by other systems. Here, we use a prefix of KEK_ for filter aliases. Adjust the instructions if a different naming convention is used.

For AWS KMS, the KMS configuration looks like this.

For TLS trust and TLS client authentication configuration, the filter accepts the same TLS parameters as Upstream TLS except the PEM store type is currently not supported.

As the administrator, use either the AWS Console or CLI to create a Symmetric key with Encrypt and decrypt usage. Multi-region keys are supported. It is not possible to make use of keys from other AWS accounts^[3].

Give the key an alias as described in Establish an aliasing convention for keys within AWS KMS.

If using the CLI, this can be done with commands like this:

Once the key is created, it is recommended to use a key rotation policy.

Note: OauthBearer config follows kafka’s properties

To intercept all Fetch Requests your class would implement FetchRequestFilter:

To intercept all Fetch Responses your class would implement FetchResponseFilter:

To intercept all Fetch Requests and all Fetch Responses your class would implement FetchRequestFilter and FetchResponseFilter:

Specific Message Filter interfaces are mutually exclusive with Request/Response. Kroxylicious will reject invalid combinations of interfaces.

The FilterContext is the factory for the FilterResult objects.

There are two convenience methods^[4] that simply allow a filter to forward a result to the next filter. We’ve already seen these in action above.

To access richer features, use the filter result builders context.requestFilterResultBuilder() and responseFilterResultBuilder().

Filter result builders allow you to:

The builder lets you combine legal behaviours together. For instance, to close the connection after forwarding a response to a client, a response filter could use:

The builders yield either a completed CompletionStage<FilterResult> which can be returned directly from the filter method, or bare FilterResult. The latter exists to support asynchronous programming styles allowing you to use your own Futures.

This is a common pattern, we want to inspect or modify a message. For example:

In some cases we may wish to not forward a request from the client to the Cluster. Instead, we want to intercept that request and generate a response message in a Kroxylicious Protocol Filter and send it towards the client. This is called a short-circuit response.

For example:

This will respond to all Create Topic requests with an error response without forwarding any of those requests to the Cluster.

Filters can make additional asynchronous requests to the Cluster. This is useful if the Filter needs additional information from the Cluster in order to know how to mutate the filtered request/response.

The Filter can make use of CompletionStage chaining features ([#thenApply() etc.) to organise for actions to be done once the asynchronous request completes. For example, it could chain an action that mutates the filtered request/response using the asynchronous response, and finally, chain an action to forward the request/response to the next filter.

The asynchronous request/response will be intercepted by Filters upstream of this Filter. Filters downstream of this Filter (and the Client) do not see the asynchronous response.

Let’s take a look at an example. We’ll send an asynchronous request towards the Cluster for topic metadata while handling a FetchRequest and use the response to mutate the FetchRequest before passing it to the next filter in the chain.

As you have read above, we need to know the API version we want our request to be encoded at. Your filter can discover what versions of an API the Kafka Cluster supports. To do this use the ApiVersionsService available from the FilterContext to determine programmatically what versions of an API are support and then write code to make a suitable request object.

You may wish to restrict your Filter to only apply to specific versions of an API. For example, "intercept all FetchRequest messages greater than api version 7". To do this you can override a method named shouldHandleXyz[Request|Response] on your filter like:

In the PortPerBroker scheme, Kroxylicious automatically opens a port for each virtual cluster’s bootstrap and one port per broker of each target cluster. So, if you have two virtual clusters, each targeting a Kafka Cluster of three brokers, Kroxylicious will bind eight ports in total.

PortPerBroker is designed to work best with simplistic configurations. It is preferable if the target cluster has sequential, stable broker ids and a known minimum broker id (like 0,1,2 for a cluster of 3 brokers). It can work with non-sequential broker ids, but you would have to expose maxBrokerId - minBrokerId ports, which could be a huge number if your cluster included broker ids 0 and 20000.

Kroxylicious monitors the broker topology of the target cluster at runtime. It will adjust the number of open ports dynamically. If another broker is added to the Kafka Cluster, Kroxylicious will automatically open a port for it. Similarly, if a broker is removed from the Kafka Cluster, Kroxylicious will automatically close the port that was assigned to it.

The PortPerBroker scheme can be used with either clear text or TLS downstream connections.

The brokerAddressPattern configuration parameter understands the replacement token $(nodeId). It is optional. If present, it will be replaced by the node.id (or broker.id) assigned to the broker of the target cluster.

For example if your configuration looks like the above and your cluster has three brokers, your Kafka Client will receive broker address information like this:

The numberOfBrokerPorts imposes a maximum on the number of brokers that a Kafka Cluster can have. Set this value to be as high as the maximum number of brokers that your operational rules allow for a Kafka Cluster.

Note that each broker’s id must be greater than or equal to lowestTargetBrokerId, and less than lowestTargetBrokerId + numberOfBrokerPorts. The current strategy for mapping node ids to ports is nodeId → brokerStartPort + nodeId - lowestTargetBrokerId. So a configuration like:

can only map broker ids 1000, 1001 and 1002 to ports 9193, 9194 and 9195 respectively. You would have to reconfigure numberOfBrokerPorts to accommodate new brokers being added to the cluster.

The original PortPerBroker scheme has the limitation that we only control the lowest target brokerId and a maximum number of brokers. We then expect all brokerIds to fall into the range [lowestBrokerId, lowestBrokerId + maxBrokerCount) We must be able to map every possible broker id to a unique port so that in a cluster of Kroxylicious proxies all members will deterministically map nodeId X to a port Y. Meaning that we may need to allocate many ports that are not required if there are large gaps between nodeIds in the target cluster.

The Range Aware Port Per Node schema introduces the idea of Node ID Ranges, allowing you to model what nodeId ranges exist in the target cluster so that the proxy can expose a more compact number of ports but still retain this deterministic mapping from nodeId to port.

Aside from how it maps nodeIds to ports it behaves the same as Port-Per-Broker.

NodeIdRanges must be distinct, a nodeId cannot be part of more than one range.

The brokerAddressPattern configuration parameter understands the replacement token $(nodeId). It is optional. If present, it will be replaced by the node.id (or broker.id) assigned to the broker of the target cluster.

For example: if I have a target cluster using KRaft that looks like:

Then we can model this as three Node Id Ranges:

Which will result in this mapping from nodeId to Port

In the SniRouting scheme, Kroxylicious uses SNI information to route traffic to either the boostrap or individual brokers. As this relies on SNI (Server Name Indication), the use of Downstream TLS is required.

With this scheme, you have the choice to share a single port across all virtual clusters, or you can assign a different port to each. Single port operation may have cost advantages when using load balancers of public clouds, as it allows a single cloud provider load balancer to be shared across all virtual clusters.

The brokerAddressPattern configuration parameters requires that the $(nodeId) token is present within it. This is replaced by the node.id (or `broker.id) assigned to the broker of the target cluster. This allows Kroxylicious to generate separate routes for each broker.

Here’s how to enable TLS for the downstream side. This means the Kafka Client will connect to the virtual cluster over TLS rather than clear text. For this, you will need to obtain a TLS certificate for the virtual cluster from your Certificate Authority.

Kroxylicious accepts key material in PKCS12 or JKS keystore format, or PEM formatted file(s). The following configuration illustrates configuration with PKCS12 keystore.

Alternatively, if your key material is in separate PEM files (private key, and certificate/intermediates), the following configuration may be used:

Here’s show to enable TLS for the upstream side. This means that Kroxylicious connects to the (physical) Kafka Cluster over TLS. For this, your Kafka Cluster must have already been configured to use TLS.

By default, Kroxylicious inherits what it trusts from the platform it is running on and uses this to determine whether the Kafka Cluster is trusted or not.

To support cases where trust must be overridden (such as use-cases involving the use of private CAs or self-signed certificates), Kroxylicious accepts override trust material in PKCS12 or JKS keystore format, or PEM formatted certificates.

The following illustrates enabling TLS, inheriting platform trust:

The following illustrates enabling TLS but with trust coming from a PKCS12 trust store instead of the platform:

The following illustrates connection to physical cluster using TLS client authentication (aka Mutual TLS).

It is also possible to disable trust so that Kroxylicious will connect to any Kafka Cluster regardless of its certificate validity.

YAML Proxy level configuration

Defaults to binding to 0.0.0.0:9190. If no endpoints are configured the admin http server is not created.

This configuration:

Prometheus is connected to the micrometer global registry so Filters can record metrics against it, and they will be available as part of the Prometheus scrape data.

If you executed curl localhost:9999/metrics you should see metrics like:

We can add common tags that will be added to all metrics. These will be available as labels from the Prometheus scrape.

To configure common tags use configuration:

Micrometer has a concept of MeterBinder:

By registering some standard binders shipped with micrometer you can expose metrics about the JVM and system which can observe JVM memory usage, garbage collection and other behaviour.

To configure multiple binders you can use configuration like:

And those named binders will be bound to the global meter registry

Filters can use the static methods of Metrics to register metrics with the global registry. Or use Metrics.globalRegistry to get a reference to the global registry. Metrics registered this way will be automatically available through the prometheus scrape endpoint.