[FLINK-29398][connector/kafka] Provide rack ID to Kafka Source to take advantage of Rack Awareness #20

jeremy-degroot · 2023-03-30T17:19:03Z

What is the purpose of the change

This PR adds a new method to KafkaSourceBuilder that sets the client.id config for the KafkaSource to the value returned by the provided Supplier. It needs to be a Supplier because it needs to run on the TaskManager, and can't be determined at Job submit time like other configs.

Brief change log

Add setRackId to KafkaSourceBuilder
Plumb rackId into KafkaPartitionSplitReader
Add rack id tests
Document RackId feature

Verifying this change

Added tests for the KafkaSplitReader that verify behaviors for null rackId Supplier, null and empty return values, and provided values.
Manually verified the change by running a 3-node cluster that covered two "racks" (AWS Availability Zones) against an Amazon MSK cluster.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? docs/Javadocs

- Added validation for the correctness of the rack ID supplied

Update docummentation to include rack awareness

Add RackId Example

remove validation session and add it to general rackID. Add client_rack documentation, Add sentence to link rackID to client.rack

FLINK-29398 Provide rack ID to Kafka Source to take advantage of Rack Awareness

boring-cyborg · 2023-03-30T17:19:07Z

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

tzulitai · 2023-04-03T15:48:14Z

Thanks for the PR @jeremy-degroot. I'll try to find some time to review this over this week.

RamanVerma · 2023-04-06T21:15:36Z

.../src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java

        this.subtaskId = context.getIndexOfSubtask();
        this.kafkaSourceReaderMetrics = kafkaSourceReaderMetrics;
        Properties consumerProps = new Properties();
        consumerProps.putAll(props);
        consumerProps.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, createConsumerClientId(props));
+        setConsumerClientRack(consumerProps, rackIdSupplier);


Can this be resolved in the caller (KafkaSource) and ConsumerConfig.CLIENT_RACK_CONFIG can be set there. The benefit would be that you do not need to change this file.

I looked into this, and while it's certainly possible to do it that way the testing path is much more complex. Since the Supplier would have to be resolved in another Supplier that's passed to the Reader, testing that the behavior is as expected in the actual execution path is difficult. In KafkaPartitionSplitReader, we can call the constructor directly and verify that the rackIdSupplier is called, and then also verify it does what we need it to by verifying the behavior of the helper method you noted further down.

Yeah KafkaSource couldn't resolve the rack since the rack may not be the same for JM and TM. I would suggest to resolve this at the KafkaSourceReader if possible and avoid this low level change.

I don't think it can be done in KafkaSourceReader either, since that doesn't interact with the Supplier at all. The alternative seems to be to resolve it in KafkaSourceFetcherManager, since that extends SingleThreadFetcherManager, where both Suppliers eventually run. But that would involve a rather messy overriding of the createSplitFetcher() method from that included class, and I don't think that's any better.

RamanVerma

thanks for the change @jeremy-degroot, I have made some minor comments

RamanVerma · 2023-04-06T21:32:12Z

.../test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java

+                        properties,
+                        UnregisteredMetricsGroup.createSourceReaderMetricGroup(),
+                        rackIdSupplier);
+        reader.setConsumerClientRack(properties, rackIdSupplier);


Supplier function should get called from KafkaPartitionSplitReader constructor. Do you need to call this method. Same for line 365

If this call is not needed, you can make the implementation of setConsumerClientRack private.

As far as I know, neither the KafkaConsumer nor the KafkaPartitionSplitReader offers any way to examine the final consumer Properties, so to test the behavior of setConsumerClientRack we have to test it directly.

You can expose a method @VisibleForTesting or use reflection?

.../test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java

mas-chen · 2023-06-16T00:45:09Z

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSource.java

@@ -98,20 +98,24 @@
    private final KafkaRecordDeserializationSchema<OUT> deserializationSchema;
    // The configurations.
    private final Properties props;
+    // Client rackId callback
+    private final Supplier<String> rackIdSupplier;


This should be a "CloseableSerializableSupplier". A supplier may contain a http client for example and also needs to be closed properly by Flink

You're right about the Serializable bit, good catch.

Is there a straightforward way to do that while still allowing for Lambdas? Maybe provide one method for the basic Supplier and another for ClosableSerializableSupplier for when you need that degree of safety.

I think we must address this before we merge this PR.
At the very least, the Supplier needs to be serializable.

You can use the SerializableSupplier interface from Flink.

docs/content/docs/connectors/datastream/kafka.md

...onnector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java

Co-authored-by: Mason Chen <[email protected]>

- Changed the name of setRackId to setRackIdSupplier

# Conflicts: # docs/content/docs/connectors/datastream/kafka.md # flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java

pzim · 2023-08-29T22:53:15Z

@jeremy-degroot - this capability is one that we are currently in need of. Any idea when this might be approved/merged/released? Or are there other alternatives we might use to achieve rack awareness in flink?

pzim · 2023-09-14T16:57:30Z

@mas-chen - any chance this will be reviewed/approved soon?

sean-scott-lr · 2023-09-19T16:16:37Z

I need this change as well, and it appears it wont be going into any release any time soon. I tried to extend the classes to add this functionality, but due to the excessive package/private access modifiers its not possible. Any suggestions on how I may do that without having to fork this project?

tzulitai

Looks like there's two remaining items to address before we can merge this:

The rack ID supplier needs to be at least Serializable. Closeable variant can probably be added later.
Move the lazy configuration of rack id on client properties to be higher in the initialization call stacks - as it is right now, it's happening at too low-level (at the split reader).

tzulitai · 2023-09-20T18:46:13Z

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSource.java

@@ -98,20 +98,24 @@
    private final KafkaRecordDeserializationSchema<OUT> deserializationSchema;
    // The configurations.
    private final Properties props;
+    // Client rackId callback
+    private final Supplier<String> rackIdSupplier;


I think we must address this before we merge this PR.
At the very least, the Supplier needs to be serializable.

You can use the SerializableSupplier interface from Flink.

… in KafkaSource - Changed the type of the KafkaSourceBuilder constructor parameter to a String - Changed the type of rackIdSupplier to a SerializableSupplier

…ry Supplier - Added test cases verifying the rack ID supplier is called at the right time - Simplified the tests in KafkaPartitionSplitReaderTest

tzulitai

Thanks @jeremy-degroot! Just two minor nitpicks, I'll address them when I merge this.

Merging now ...

tzulitai · 2023-09-29T15:58:01Z

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSource.java


    KafkaSource(
            KafkaSubscriber subscriber,
            OffsetsInitializer startingOffsetsInitializer,
            @Nullable OffsetsInitializer stoppingOffsetsInitializer,
            Boundedness boundedness,
            KafkaRecordDeserializationSchema<OUT> deserializationSchema,
-            Properties props) {
+            Properties props,
+            SerializableSupplier<String> rackIdSupplier) {


Suggested change

SerializableSupplier<String> rackIdSupplier) {

@Nullable SerializableSupplier<String> rackIdSupplier) {

tzulitai · 2023-09-29T17:37:42Z

...onnector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java

+    public KafkaSourceBuilder<OUT> setRackIdSupplier(SerializableSupplier<String> rackIdCallback) {
+        this.rackIdSupplier = rackIdCallback;


nit: can we unify the naming here to just refer to this as the rackIdSupplier?

tzulitai · 2023-09-29T17:53:33Z

Actually, @jeremy-degroot would you be able to squash all your changes into one commit that is rebased on latest main branch? That would be super helpful.

…k Awareness Co-authored-by: jcmejias1 <[email protected]> Co-authored-by: Mason Chen <[email protected]> Co-authored-by: Ethan Gouty <[email protected]> Co-authored-by: Jeremy DeGroot <[email protected]> Co-authored-by: Siva Venkat Gogineni <[email protected]> This closes apache#53. This closes apache#20.

…k Awareness This closes apache#53. This closes apache#20. Co-authored-by: Jeremy DeGroot <[email protected]> Co-authored-by: jcmejias1 <[email protected]> Co-authored-by: Mason Chen <[email protected]> Co-authored-by: Ethan Gouty <[email protected]> Co-authored-by: Siva Venkat Gogineni <[email protected]>

GoutyEthan and others added 12 commits March 30, 2023 11:32

Add rackIdSupplier and funnel into KafkaPartitionSplitReader

9c23f7d

Add rackIdSupplier test to KafkaPartitionSplitReaderTest

b4aa761

- Formatting

83771ec

- Added validation for the correctness of the rack ID supplied

Replaced the literals with a variable

190e2e6

Changed up handling for the values around the rack ID

7c2cf26

Update kafka.md

cf218ad

Update docummentation to include rack awareness

Update kafka.md

ba451d9

Update docummentation to include rack awareness

Update kafka.md

d85ecd1

Add RackId Example

Update kafka.md

2a6aaa5

remove validation session and add it to general rackID. Add client_rack documentation, Add sentence to link rackID to client.rack

Added documentation for the setConsumerClientRack private method.

c6550c9

spotlessApply

167098e

Merge pull request #1 from jeremy-degroot/FLINK-29398-rack-awareness

5a7aca7

FLINK-29398 Provide rack ID to Kafka Source to take advantage of Rack Awareness

boring-cyborg bot added component=Connectors/Kafka component=Documentation labels Mar 30, 2023

jeremy-degroot mentioned this pull request Mar 30, 2023

FLINK-29398 [Kafka Source] Provide rack ID to Kafka Source to take advantage of Rack Awareness apache/flink#22309

Closed

jeremy-degroot changed the title ~~FLINK-29398 Provide rack ID to Kafka Source to take advantage of Rack Awareness~~ FLINK-29398 [connectors/kafka] Provide rack ID to Kafka Source to take advantage of Rack Awareness Mar 30, 2023

jeremy-degroot changed the title ~~FLINK-29398 [connectors/kafka] Provide rack ID to Kafka Source to take advantage of Rack Awareness~~ FLINK-29398 [connector/kafka] Provide rack ID to Kafka Source to take advantage of Rack Awareness Mar 30, 2023

jeremy-degroot changed the title ~~FLINK-29398 [connector/kafka] Provide rack ID to Kafka Source to take advantage of Rack Awareness~~ [FLINK-29398][connector/kafka] Provide rack ID to Kafka Source to take advantage of Rack Awareness Mar 30, 2023

tzulitai self-requested a review April 3, 2023 15:47

RamanVerma reviewed Apr 6, 2023

View reviewed changes

jeremy-degroot added 2 commits April 7, 2023 12:47

Combined a couple of similar tests

a829ecf

Combined a couple of similar tests

c98b9f3

jeremy-degroot requested a review from RamanVerma April 11, 2023 14:22

mas-chen reviewed Jun 16, 2023

View reviewed changes

docs/content/docs/connectors/datastream/kafka.md Outdated Show resolved Hide resolved

mas-chen reviewed Jun 16, 2023

View reviewed changes

docs/content/docs/connectors/datastream/kafka.md Outdated Show resolved Hide resolved

mas-chen reviewed Jun 16, 2023

View reviewed changes

docs/content/docs/connectors/datastream/kafka.md Outdated Show resolved Hide resolved

mas-chen reviewed Jun 16, 2023

View reviewed changes

...onnector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java Outdated Show resolved Hide resolved

jeremy-degroot and others added 4 commits June 30, 2023 09:25

Update docs/content/docs/connectors/datastream/kafka.md

be13242

Co-authored-by: Mason Chen <[email protected]>

- Cleaned up documentation

342cd72

- Changed the name of setRackId to setRackIdSupplier

Merge branch 'apache:main' into main

d0b24f9

Merge branch 'FLINK-29398-rack-awareness' into main

9acf6ba

# Conflicts: # docs/content/docs/connectors/datastream/kafka.md # flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java

jeremy-degroot requested a review from mas-chen August 30, 2023 16:45

Merge branch 'apache:main' into main

91cfa27

Merge branch 'apache:main' into main

3cf0794

tzulitai reviewed Sep 20, 2023

View reviewed changes

jeremy-degroot added 2 commits September 21, 2023 11:36

- Moved the resolution of the rackId supplier into the readerSupplier…

b0d5d02

… in KafkaSource - Changed the type of the KafkaSourceBuilder constructor parameter to a String - Changed the type of rackIdSupplier to a SerializableSupplier

- Added handling for a null rackIdSupplier in the source reader facto…

e3bd8bc

…ry Supplier - Added test cases verifying the rack ID supplier is called at the right time - Simplified the tests in KafkaPartitionSplitReaderTest

tzulitai approved these changes Sep 29, 2023

View reviewed changes

jeremy-degroot closed this Sep 29, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-29398][connector/kafka] Provide rack ID to Kafka Source to take advantage of Rack Awareness #20

[FLINK-29398][connector/kafka] Provide rack ID to Kafka Source to take advantage of Rack Awareness #20

jeremy-degroot commented Mar 30, 2023

boring-cyborg bot commented Mar 30, 2023

tzulitai commented Apr 3, 2023

RamanVerma Apr 6, 2023

jeremy-degroot Apr 7, 2023

mas-chen Jun 16, 2023

jeremy-degroot Jun 30, 2023

RamanVerma left a comment

RamanVerma Apr 6, 2023

RamanVerma Apr 6, 2023

jeremy-degroot Apr 7, 2023

mas-chen Jun 16, 2023

mas-chen Jun 16, 2023 •

edited

Loading

jeremy-degroot Jun 30, 2023

tzulitai Sep 20, 2023

pzim commented Aug 29, 2023

pzim commented Sep 14, 2023

sean-scott-lr commented Sep 19, 2023

tzulitai left a comment •

edited

Loading

tzulitai Sep 20, 2023

tzulitai left a comment

tzulitai Sep 29, 2023

tzulitai Sep 29, 2023

tzulitai commented Sep 29, 2023

	SerializableSupplier<String> rackIdSupplier) {
	@Nullable SerializableSupplier<String> rackIdSupplier) {

		public KafkaSourceBuilder<OUT> setRackIdSupplier(SerializableSupplier<String> rackIdCallback) {
		this.rackIdSupplier = rackIdCallback;

[FLINK-29398][connector/kafka] Provide rack ID to Kafka Source to take advantage of Rack Awareness #20

[FLINK-29398][connector/kafka] Provide rack ID to Kafka Source to take advantage of Rack Awareness #20

Conversation

jeremy-degroot commented Mar 30, 2023

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

boring-cyborg bot commented Mar 30, 2023

tzulitai commented Apr 3, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RamanVerma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mas-chen Jun 16, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pzim commented Aug 29, 2023

pzim commented Sep 14, 2023

sean-scott-lr commented Sep 19, 2023

tzulitai left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tzulitai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tzulitai commented Sep 29, 2023

mas-chen Jun 16, 2023 •

edited

Loading

tzulitai left a comment •

edited

Loading