Package io.kroxylicious.proxy.filter

@ReturnValuesAreNonnullByDefault @DefaultAnnotationForParameters(edu.umd.cs.findbugs.annotations.NonNull.class) @DefaultAnnotation(edu.umd.cs.findbugs.annotations.NonNull.class) package io.kroxylicious.proxy.filter

The protocol filter API.

An interface is provided for each kind of request and response in the Kafka protocol, e.g. ProduceRequestFilter. Protocol Filter implementations inherit whichever of the per-RPC interfaces they need to intercept. They can inherit multiple interfaces if necessary. For filters which needs to intercept most or all of the protocol it is more convenient to inherit RequestFilter and/or ResponseFilter.

Important facts about the Kafka protocol

Pipelining

The Kafka protocol supports pipelining (meaning a client can send multiple requests, before getting a response for any of them). Therefore when writing a filter implementation do not assume you won't see multiple requests before seeing any corresponding responses.

Ordering

A broker does not, in general, send responses in the same order as it receives requests. Therefore when writing a filter implementation do not assume ordering.

Local view

A client may obtain information from one broker in a cluster and use it to interact with other brokers in the cluster (or the same broker, but on a different connection, and therefore a different channel and filter chain). A classic example would be a producer or consumer making a metadata connection and Metadata request to a broker and then connecting to a partition leader to producer/consume records (Produce and Fetch requests).

So although your filter implementation might intercept both Metadata and Produce request/response (for example), those requests will not pass through the same instance of your filter implementation. Therefore it is incorrect, in general, to assume your filter has a global view of the communication between the client and broker.

Implementing Filters

Filter Results

Filter implementation must return a CompletionStage containing a FilterResult object. It is the job of FilterResult to convey what message is to be forwarded to the next filter in the chain (or client/broker if at the chain's beginning or end). It is also used to carry instructions such as indicating that the connection must be closed, or a message dropped.

If the filter returns a CompletionStage that is already completed normally, Kroxylicious will immediately perform the action described by the FilterResult.

If the CompletionStage completes exceptionally, the connection is closed. This also applies if the CompletionStage does not complete within a timeout (20000 milliseconds).

Deferring Forwards

The filter may return a CompletionStage that is not yet completed. When this happens, Kroxylicious will pause reading from the downstream (the Client writes will eventually block), and it begins to queue up in-flight requests/responses arriving at the filter. This is done so that message order is maintained. Once the CompletionStage completes, the action described by the FilterResult is performed, reading from the downstream resumes and any queued up requests/responses are processed.

IMPORTANT: The pausing of reads from the downstream is a relatively costly operation. To maintain optimal performance filter implementations should minimise the occasions on which an incomplete CompletionStage is returned.

Creating Filter Result objects

The FilterContext is the factory for the FilterResult objects.

There are two convenience methods that simply allow a filter to immediately forward a result:

To access richer features, use the filter result builders:

Thread Safety

The Filter API provides the following thread-safety guarantees:

  1. There is a single thread associated with each connection and this association lasts for the lifetime of connection..
  2. Each filter instance is associated with exactly one connection.
  3. Construction of the filter instance and dispatch of the filter methods onXxxRequest and onXxxResponse takes place on that same thread.
  4. Any computation stages chained to the CompletionStage returned by FilterContext.sendRequest(org.apache.kafka.common.message.RequestHeaderData, org.apache.kafka.common.protocol.ApiMessage) using the default execution methods (using methods without the suffix async) or default asynchronous execution (using methods with suffix async that employ the stage's default asynchronous execution facility) are guaranteed to be performed by that same thread. Computation stages chained using custom asynchronous execution (using methods with suffix async that take an Executor argument) do not get this guarantee.

Filter implementations are free to rely on these guarantees to safely maintain state within fields of the Filter without employing additional synchronization.

If a Filter needs to do some asynchronous work and mutate members of the Filter, they can access the thread for the connection via FilterFactoryContext.filterDispatchExecutor(), which is available in FilterFactory.createFilter(FilterFactoryContext, java.lang.Object) when it is creating an instance of a Filter. Ensure that any work executes quickly as this is the IO thread for potentially many connections.