KAFKA-14365: Extract common logic from Fetcher #13425
Conversation
|
|
Extract logic from Fetcher into AbstractFetcher. Also introduce FetchConfig as a more concise way to delineate state from incoming configuration. Formalized the defaults in CommonClientConfigs and ConsumerConfig to be accessible elsewhere.
|
@guozhangwang @hachikuji @rajinisivaram @philipnee This is ready for a review from whomever has the time. |
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
Show resolved
Hide resolved
| @@ -797,24 +799,15 @@ public KafkaConsumer(Map<String, Object> configs, | |||
| config.getBoolean(ConsumerConfig.THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED), | |||
| config.getString(ConsumerConfig.CLIENT_RACK_CONFIG)); | |||
| } | |||
| FetchConfig<K, V> fetchConfig = new FetchConfig<>(config, keyDeserializer, valueDeserializer, isolationLevel); | |||
There was a problem hiding this comment.
This is a nice improvement, +1!
There was a problem hiding this comment.
| FetchConfig<K, V> fetchConfig = new FetchConfig<>(config, keyDeserializer, valueDeserializer, isolationLevel); | |
| FetchConfig<K, V> fetchConfig = new FetchConfig<>(config, this.keyDeserializer, this.valueDeserializer, isolationLevel); |
There was a problem hiding this comment.
@kirktrue , these should be this.keyDeserializer and this.valueDeserializer. Otherwise the original values may be null and will cause an NPE on line 300 of CompletedFetch.java
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
Outdated
Show resolved
Hide resolved
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java
Show resolved
Hide resolved
| * | ||
| * @return true if there are completed fetches, false otherwise | ||
| */ | ||
| boolean hasCompletedFetches() { |
There was a problem hiding this comment.
I made a quick browse on this class assuming that all the logic are not changed, i.e. mostly copy-paste here except the FetchConfig wrapping. Please LMK otherwise and I will give it a deeper look.
There was a problem hiding this comment.
I intentionally did my best to keep the code I moved from Fetcher into AbstractFetch identical. There was some minor refactoring which could technically introduce bugs.
|
I've checked the failed tests, which are irrelevant. |
|
LGTM, merged to trunk. |
|
@guozhangwang , @kirktrue , this PR is causing NPEs downstream. I've added a comment above that indicates the problematic lines. Thanks. |
The
Fetcherclass is used internally by theKafkaConsumerto fetch records from the brokers. There is ongoing work to create a new consumer implementation with a significantly refactored threading model. The threading refactor work requires a similarly refactoredFetcher.This task includes refactoring
Fetcherby extracting out some common logic to allow forthcoming implementations to leverage it.FetcherintoAbstractFetchFetchConfigas a concise way to delineate incoming configuration from stateCommonClientConfigsandConsumerConfigto be accessible elsewhereCommitter Checklist (excluded from commit message)