How to dynamically forward topics using header data

[lework]

Requirement:

Forward messages to the corresponding topic based on the value of the message header tag field.

my_topic ==> dev_my_topic

echo "hello there" | kcat -b localhost:9192 -P -t my_topic -H "tag=dev" -d all 

code

    @Override
    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, RequestHeaderData header, ProduceRequestData request, FilterContext context) {
        if (!config.isEnabled()) {
            return context.forwardRequest(header, request);
        }

        request.topicData().forEach(topic -> setTopicName(context, topic::setName, topic));
        return context.forwardRequest(header, request);
    }

    @Override
    public CompletionStage<ResponseFilterResult> onProduceResponse(short apiVersion, ResponseHeaderData header, ProduceResponseData response,
                                                                   FilterContext context) {
        response.responses().forEach(topic -> reductionTopicName(context, topic::setName, topic));
        return context.forwardResponse(header, response);
    }

The problem is that

kcat keeps sending messages in a loop, but dev_my_topic has no messages. What else needs to be done to forward messages to dev_my_topic

[robobario]

Hi @lework thanks for the issue

The proxy currently can not easily route to a different topic based on message content.

With the current model:

1. the proxy receives a connection from the client addressing a specific broker (by port or TLS SNI name)
2. establishes a connection to the relevant target upstream broker
3. forwards requests/responses between them.

Kafka Clients are aware of which broker is the leader for a topic/partition and expect that they can send records to a specific broker.

So, the kafka client will have obtained metadata for topic my_topic and knows the leader broker for each partition. It sends records for partition X to the leader broker for that partition.

It could be the case that:

• dev_my_topic has a different partition count to my_topic
• the partitions of dev_my_topic live on different brokers (compared to the partitions of my_topic), with different leader brokers.

So when we forward the produce request on to the broker (the leader for my_topic partition X) if we modify the topic name, we may get an error from the broker saying it is not the leader for that topic/partition.

Being able to route messages based on content would be a very powerful feature, but means the Proxy has to be aware of the topic metadata and be able to forward messages to the correct leader broker for each partition, potentially fanning out if a single ProduceRequest contains multiple partitions that reside on different brokers. Which has further complications that the proxy currently relies on the client to drive SASL authentication, we'd probably need the proxy to be able to drive it's own SASL negotiations.

Maybe if you try producing with the java bin/kafka-console-producer.sh it will log some errors out.

What we have been able to do in other Filters like the MultiTenantFilter is present a physical kafka topic like dev_my_topic to clients as a topic with a different name like my_topic. We do this by intercepting the Metadata returned to clients and changing the topic names there. Virtualizing the resource so it appears to clients that the topic is named my_topic, but then the ProduceRequestFilter will update the name to dev_my_topic. But this is not done based on the message content.

We are currently looking at making a Filter that does this kind of namespacing based on the SASL principal, though we haven't got topic names in scope initially.

[robobario]

Would be great to know more about your use case. What is your goal in routing by header?

[lework]

Thank you for your reply.

This is due to an abtest scenario requirement. Data a needs to be processed by the logic of consumer a, and data b needs to be processed by the logic of consumer b.

Therefore, my_topic and dev my_topic configurations are exactly the same.

I've done some more testing and found that if I intercept the onMetadataRequest and onMetadataResponse calls and specify the topic name directly, Kcat can successfully send the message. However, since this prevents me from using the header data to determine if the message is being sent, I'd like to know if there are any other solutions.

[k-wall]

Thank you for your reply.

This is due to an abtest scenario requirement. Data a needs to be processed by the logic of consumer a, and data b needs to be processed by the logic of consumer b.

Therefore, my_topic and dev my_topic configurations are exactly the same.

I think even if topic configurations are the same (number of partitions), you won't have certainty that the leaders for partitions are on the same brokers. Topic A partition 1 leader might be on broker 0. Topic B partition 1 leader might be on broker 4. If you rewrite the produce request, swapping topic name A for B, broker 0 will reject the batch because it is not leader for Topic B partition 1.

I've done some more testing and found that if I intercept the onMetadataRequest and onMetadataResponse calls and specify the topic name directly, Kcat can successfully send the message. However, since this prevents me from using the header data to determine if the message is being sent, I'd like to know if there are any other solutions.

I'm not following what you mean. I don't think intercepting metadata request/response helps. It is fundamental to Kafka that producing clients connect to topic partition leaders.

I wonder if the Routing API https://github.com/kroxylicious/design/pull/70/files (proposal) would help your use-case (in the long term).

I can think of a "hacky" solution which should work with kroxylicious as it stands today. I haven't tried it.

Have your ProduceRequestFiler#onProduceRequest use the Kafka Producer API directly. Write code in the filter in that connects to your Kafka Cluster using a Producer and then use org.apache.kafka.clients.producer.Producer#send() to send the records. The Kafka Client will connect to the brokers it needs to. Use the send future completion to trigger a "short-circuit" response back to the client crafting the ProduceResponseData so the client knows the outcome.

[lework]

Thank you for your reply.

I'm not following what you mean. I don't think intercepting metadata request/response helps. It is fundamental to Kafka that producing clients connect to topic partition leaders.

The processing I performed can be understood as consistent with the logic in the MultiTenantFilter.

https://github.com/kroxylicious/design/pull/70/files

This feature looks great. I wonder if it will support routing based on message data in the future.

Have your ProduceRequestFiler#onProduceRequest use the Kafka Producer API directly. Write code in the filter in that connects to your Kafka Cluster using a Producer and then use org.apache.kafka.clients.producer.Producer#send() to send the records. The Kafka Client will connect to the brokers it needs to. Use the send future completion to trigger a "short-circuit" response back to the client crafting the ProduceResponseData so the client knows the outcome.

This is a great approach. Does this mean the same message is sent to both my_topic and dev_my_topic? I'll test this idea later.

However, I also have an approach to address my current scenario.

Producer program: Implement dynamic routing for message sending (my_topic → dev_my_topic)

kroxylicious: When no consumer program exists for dev_my_topic, restore dev_my_topic to my_topic, allowing my_topic's consumers to process the data.

Why not use the producer program to implement kroxylicious functionality? Because there are too many producer programs, and they are written in different languages. Handling kroxylicious functionality would be overly complex and would hinder future unified updates.

Data Flow

user_data(a) ===> Producer Program (my_topic) ==> kroxylicious (my_topic) ==> Consumer Program a (my_topic)

user_data(b) ===> Producer Program (dev_my_topic) ==> kroxylicious (dev_my_topic) ==> Consumer Program b (dev_my_topic)

When consumer program b shuts down

user_data(b) ===> producer program (dev_my_topic) ==> kroxylicious(my_topic) ==> consumer program a (my_topic)

[tombentley]

Moving to a discussion