Hi All, We're seeing that Flink consumers are one of the main reasons for a spike in our AWS data transfer costs. We'd like to use a rack-aware Flink consumer client so we can reduce these costs. Is it currently possible in Flink? I see this ticket: https://issues.apache.org/jira/browse/FLINK-29398 and its still in open state. There is a FLIP page as well for the same. Could someone please clarify on this? Thank you! 🙂

Hi, I actually created that. We had the same problem in my team, we're hoping to get to this sometime this quarter. Scheduling time to work on open source projects, even ones that benefit us directly, is challenging at best.

Hi @Jeremy DeGroot - thanks for responding! That's great. Yes, we definitely need this. Please let me know if there's anything I could do - we can try and get an initial design out.

We'll be working on it at some point. We implemented a version of it on the new KafkaSource API, but it was not quite ready for production at that time so we implemented it AGAIN on the old KafkaConsumer API. We still need to do it for the new API because the old one is deprecated and I think going away in 1.16 or 1.17, so if we don't have an implementation up we'll be back to getting killed by bandwidth costs

Thanks @Jeremy DeGroot! For FlinkKafkaConsumer is it that easy to pass on the consumer property

client.rack=use1-az1

and flink consumer knows to consume from topic replicas in az1? we are still using flink 1.14.

It is if your cluster isn't in multiple AZs.

The kafka cluster is in 3 different AZs. If we make sure the flink job’s all task managers nodes run on the same AZ’s (say az1) and make sure we pass

client.rack=use1-az1

would that help ?

Yes, that would work

@Jeremy DeGroot did you get a chance to work on the rack-aware flink consumer? The above strategy has a major drawback and have described here: https://github.com/spotify/flink-on-k8s-operator/issues/683 and https://stackoverflow.com/questions/76223749/flink-1-14-3-rack-aware-flink-task-manager

@Guruguha Marur Sreenivasa we've got a PR up and it's going through review.

Great, thanks @Jeremy DeGroot! When can we expect it to be in a release? Also, from which version of Flink would this feature be in?

That's up to the maintainers. I can tell you it won't be in anything before 1.18

Couple of questions on the PR: 1. I was trying to implement this in a similar way but got stuck at this point. How do you get the AZ the task manager is running in at runtime? Our Flink cluster is setup to be in application mode. So, the job submitter is the one creating the DAG and init-ing the

KafkaSource

and when I tried to get the AZ of TM at runtime, it basically was the job submitter's AZ that we got. 2. The Kafka consumer's

client.rack

should still be passed though, right? Or will it be taken care by the

KafkaPartitionSplitReader

Yeah, coming up with a

Supplier

is something I need help with. I have tried the following: 1. Tried to pass in AZ info as part of Flink operator definition. This didn't work for some reason - although I expected it to work. 2. Tried the above route to get AZ info at runtime - this was expected as the AZ we get is for the current pod and not TM

So, that's very environment and implementation-specific. For us, in EKS, we specify we've got to have capacity in each of the AZs we use. In that Pod specification, we also put the AZ in an environment variable which is simply returned by the Supplier.

I tried this exact thing. As part of TaskManager spec in the Spotify k8s Flink operator (similar to #1). Do you have an example of yours that I can refer to?

I don't know anything about the Spotify operator, sorry.

No worries. An EKS spec example should help too - I can use it as reference and come up with a CRD for our use-case.

We rolled our own deployment YAML, and from there we use a

nodeSelector

which is an EKS thing as I understand:

nodeSelector:
        "<http://topology.kubernetes.io/zone|topology.kubernetes.io/zone>": {{ quote $label }}
...
        env:
        - name: TM_NODE_AZ
          value: {{ quote $zone }}

We have in our config a mapping of

$label

to

$zone

that takes the availability zones EKS is aware of (

us-east-1[a-e]

) and maps them to the zones MSK uses (

use1-az[1-6]

). If your Kafka install is different, you may not need that. https://archive.eksworkshop.com/beginner/140_assigning_pods/node_selector/ https://docs.aws.amazon.com/ram/latest/userguide/working-with-az-ids.html

Yeah, we are using this currently as follows:

taskManager:
    nodeSelector:
      <http://topology.kubernetes.io/zone|topology.kubernetes.io/zone>: us-east-1b

In yours, its a variable,

$zone

- how is this populated?

We iterate over our label -> zone mapping

Yeah that's where I kinda got stuck. I was going to try the `KafkaPartitionSplitReader`Since the

Supplier<>

way you have is a callback and will run on the task manager than on the job submitter, your approach works. The way I was trying before is to set the

client.rack

during the

KafkaSource

init. I found a way to get the AZ a pod is running on at runtime and tried to set it to the

client.rack

variable. The problem with this is, the

KafkaSource

init happens in the job submitter and the AZ I got at runtime was of the job submitter and not the TaskManager. 😞

Yeah, there's no way to do it at Job submit time. Hence the change my team implemented.

Yup - makes sense. Waiting for your PR to be merged! Thanks for your help.