Dynamic Kafka Properties

[Salva Alcántara <sa...@gmail.com>]

I've recently started to enable rack awareness in my sources following this:

https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/kafka/#rackid

E.g.,

```java

.setRackIdSupplier(() -> System.getenv("TM_NODE_AZ"))

```

This approach allows to decouple the AZ between jobmanager & taskamangers.

There are indeed some Kafka-compatible solutions like WarpStream which
support rack awareness for producers, too. E.g.,

https://docs.warpstream.com/warpstream/byoc/configure-kafka-client/configuring-kafka-client-id-features#warpstream_az

The problem is that they pass the rack id using a normal producer property.
Because the Kafka sink builder gets the properties and serializes them in
the jobmanager, all the taskmanagers will use the same AZ as that of the
jobmanager.

Is there a way to easily pass "dynamic" properties so to say?

Would it make sense to consider a supplier-based override for the property
setters of the Kafka builders?

Regards,

Salva

Re: Dynamic Kafka Properties

[Salva Alcántara <sa...@gmail.com>]

By looking at:

https://github.com/apache/flink-connector-kafka/pull/20

where the "setRackIdSupplier" setter for the kafka source builder was
introduced, I guess one could take a more general approach and instead
consider overrides for:

"setProperty"
"setProperties"

which receive a supplier for the property/properties (this could be done
for sinks, too).

These new methods could be used for any property that one requires to be
inferred/run in the taskmanagers for what it's worth, not just the rack id
one.

I guess as of now the rack id is probably the only property requiring lazy
evaluation so to say, but in the future who knows...

Regards,

Salva

On Mon, Jan 27, 2025 at 6:11 PM Salva Alcántara <sa...@gmail.com>
wrote:

> I've recently started to enable rack awareness in my sources following
> this:
>
>
> https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/kafka/#rackid
>
> E.g.,
>
> ```java
>
> .setRackIdSupplier(() -> System.getenv("TM_NODE_AZ"))
>
> ```
>
> This approach allows to decouple the AZ between jobmanager & taskamangers.
>
> There are indeed some Kafka-compatible solutions like WarpStream which
> support rack awareness for producers, too. E.g.,
>
>
> https://docs.warpstream.com/warpstream/byoc/configure-kafka-client/configuring-kafka-client-id-features#warpstream_az
>
> The problem is that they pass the rack id using a normal producer
> property. Because the Kafka sink builder gets the properties and serializes
> them in the jobmanager, all the taskmanagers will use the same AZ as that
> of the jobmanager.
>
> Is there a way to easily pass "dynamic" properties so to say?
>
> Would it make sense to consider a supplier-based override for the property
> setters of the Kafka builders?
>
> Regards,
>
> Salva
>
>