recreate producers and consumers during config reload command #123
Conversation
kehuum
requested review from
radai-rosenblatt and
jonlee2
and removed request for
radai-rosenblatt
|
|
||
| if (recreateSucceeded) { | ||
| // replace _consumers with newly created consumers | ||
| _consumers.clear(); |
There was a problem hiding this comment.
How about moving this to the try block and move the else block into the catch block?
| _commonConsumerConfigs = new LiKafkaConsumerConfig(originalConfigs); | ||
|
|
||
| for (ClusterConsumerPair<K, V> entry : _consumers) { | ||
| getOrCreatePerClusterConsumer(newConsumers, entry.getCluster()); |
There was a problem hiding this comment.
It does what we want, but it is not obvious that newConsumers is updated. Can we add a new method called createPerClusterConsumer() and use it here and also refactor getOrCreatePerClusterConsumer()?
| convertedConfig.put(entry.getKey(), String.valueOf(entry.getValue())); | ||
| } | ||
| // TODO: add a flag in RELOAD_CONFIG_RESPONSE message to indicate result of config reload | ||
| // TODO: add a flag in RELOAD_CONFIG_RESPONSE message to indicate result of config reload |
| } | ||
|
|
||
| if (recreateSucceeded) { | ||
| // replace _producers with newly created producers |
There was a problem hiding this comment.
All comments for consumer config reload apply here as well.
| LiKafkaConsumer<K, V> curConsumer = createPerClusterConsumer(entry.getCluster()); | ||
| newConsumers.add(new ClusterConsumerPair<K, V>(entry.getCluster(), curConsumer)); | ||
| } | ||
|
|
There was a problem hiding this comment.
Actually, we will need to re-assign/subscribe the previous assignment/subscriptions with the newly created consumers.
...-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaFederatedConsumerImpl.java
Show resolved
Hide resolved
...-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaFederatedConsumerImpl.java
Outdated
Show resolved
Hide resolved
| _closed = false; | ||
|
|
||
| // keep previous partition assignment after config reload | ||
| // TODO: after subscribe() is supported, we need to also keep the original subscription | ||
| Map<ClusterDescriptor, Set<TopicPartition>> perClusterTopicPartitionSet = getPerClusterTopicPartitionSet(curPartitions); |
There was a problem hiding this comment.
Similar comment as above. Why don't you use a map from cluster descriptor to topic partition set to begin with by calling per-cluster consumer's assignment()?
Re-create producers and consumers as part of config reload command instead of lazy construction to avoid delay of errors and user impact.