[improve][client] Optimize code for MultiTopicsConsumerImpl
@@ -929,15 +929,15 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
"Should have only 1 topic for partitioned consumer");
// get topic name, then remove it from conf, so constructor will create a consumer with no topic.
- ConsumerConfigurationData cloneConf = conf.clone();
+ ConsumerConfigurationData<T> cloneConf = conf.clone();
String topicName = cloneConf.getSingleTopic();
cloneConf.getTopicNames().remove(topicName);
- CompletableFuture<Consumer> future = new CompletableFuture<>();
- MultiTopicsConsumerImpl consumer = new MultiTopicsConsumerImpl(client, topicName, cloneConf, executorProvider,
- future, schema, interceptors, true /* createTopicIfDoesNotExist */);
+ CompletableFuture<Consumer<T>> future = new CompletableFuture<>();
+ MultiTopicsConsumerImpl<T> consumer = new MultiTopicsConsumerImpl<T>(client, topicName, cloneConf,
+ executorProvider, future, schema, interceptors, true /* createTopicIfDoesNotExist */);
- future.thenCompose(c -> ((MultiTopicsConsumerImpl) c).subscribeAsync(topicName, numPartitions))
+ future.thenCompose(c -> ((MultiTopicsConsumerImpl<T>) c).subscribeAsync(topicName, numPartitions))
.thenRun(()-> subscribeFuture.complete(consumer))
.exceptionally(e -> {
log.warn("Failed subscription for createPartitionedConsumer: {} {}, e:{}",
