skaundinya15 commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r659104126
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/DeleteConsumerGroupsResult.java
##########
@@ -40,7 +43,11 @@
* individual deletions.
*/
public Map<String, KafkaFuture<Void>> deletedGroups() {
- return futures;
+ Map<String, KafkaFuture<Void>> deletedGroups = new
HashMap<>(futures.size());
+ futures.forEach((key, future) -> {
+ deletedGroups.put(key.idValue, future);
+ });
Review comment:
Nit: I think we could simplify this lambda expression further to just:
```java
futures.forEach((key, future) -> deletedGroups.put(key.idValue, future));
```
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/DescribeConsumerGroupsResult.java
##########
@@ -34,39 +35,40 @@
@InterfaceStability.Evolving
public class DescribeConsumerGroupsResult {
- private final Map<String, KafkaFuture<ConsumerGroupDescription>> futures;
+ private final Map<CoordinatorKey,
KafkaFutureImpl<ConsumerGroupDescription>> futures;
- public DescribeConsumerGroupsResult(final Map<String,
KafkaFuture<ConsumerGroupDescription>> futures) {
+ public DescribeConsumerGroupsResult(final Map<CoordinatorKey,
KafkaFutureImpl<ConsumerGroupDescription>> futures) {
this.futures = futures;
}
/**
* Return a map from group id to futures which yield group descriptions.
*/
public Map<String, KafkaFuture<ConsumerGroupDescription>>
describedGroups() {
- return futures;
+ Map<String, KafkaFuture<ConsumerGroupDescription>> describedGroups =
new HashMap<>();
+ for (Map.Entry<CoordinatorKey,
KafkaFutureImpl<ConsumerGroupDescription>> entry : futures.entrySet()) {
+ describedGroups.put(entry.getKey().idValue, entry.getValue());
+ }
Review comment:
Nit: Similar to how it's done in `DeleteConsumerGroupsResult`, perhaps
we can replace this with a lambda expression instead?
```java
futures.forEach((key, future) -> describedGroups.put(key.idValue, future));
```
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/DescribeConsumerGroupsResult.java
##########
@@ -34,39 +35,40 @@
@InterfaceStability.Evolving
public class DescribeConsumerGroupsResult {
- private final Map<String, KafkaFuture<ConsumerGroupDescription>> futures;
+ private final Map<CoordinatorKey,
KafkaFutureImpl<ConsumerGroupDescription>> futures;
- public DescribeConsumerGroupsResult(final Map<String,
KafkaFuture<ConsumerGroupDescription>> futures) {
+ public DescribeConsumerGroupsResult(final Map<CoordinatorKey,
KafkaFutureImpl<ConsumerGroupDescription>> futures) {
this.futures = futures;
}
/**
* Return a map from group id to futures which yield group descriptions.
*/
public Map<String, KafkaFuture<ConsumerGroupDescription>>
describedGroups() {
- return futures;
+ Map<String, KafkaFuture<ConsumerGroupDescription>> describedGroups =
new HashMap<>();
+ for (Map.Entry<CoordinatorKey,
KafkaFutureImpl<ConsumerGroupDescription>> entry : futures.entrySet()) {
+ describedGroups.put(entry.getKey().idValue, entry.getValue());
+ }
+ return describedGroups;
}
/**
* Return a future which yields all ConsumerGroupDescription objects, if
all the describes succeed.
*/
public KafkaFuture<Map<String, ConsumerGroupDescription>> all() {
return KafkaFuture.allOf(futures.values().toArray(new
KafkaFuture[0])).thenApply(
- new KafkaFuture.BaseFunction<Void, Map<String,
ConsumerGroupDescription>>() {
- @Override
- public Map<String, ConsumerGroupDescription> apply(Void v) {
- try {
- Map<String, ConsumerGroupDescription> descriptions =
new HashMap<>(futures.size());
- for (Map.Entry<String,
KafkaFuture<ConsumerGroupDescription>> entry : futures.entrySet()) {
- descriptions.put(entry.getKey(),
entry.getValue().get());
- }
- return descriptions;
- } catch (InterruptedException | ExecutionException e) {
- // This should be unreachable, since the
KafkaFuture#allOf already ensured
- // that all of the futures completed successfully.
- throw new RuntimeException(e);
+ nil -> {
+ Map<String, ConsumerGroupDescription> descriptions = new
HashMap<>(futures.size());
+ try {
+ for (Map.Entry<CoordinatorKey,
KafkaFutureImpl<ConsumerGroupDescription>> entry : futures.entrySet()) {
Review comment:
Nit: We could also use the lambda expression instead:
```java
public KafkaFuture<Map<String, ConsumerGroupDescription>> all() {
return KafkaFuture.allOf(futures.values().toArray(new
KafkaFuture[0])).thenApply(
nil -> {
Map<String, ConsumerGroupDescription> descriptions = new
HashMap<>(futures.size());
futures.forEach((key, future) -> {
try {
descriptions.put(key.idValue, future.get());
} catch (InterruptedException | ExecutionException e) {
// This should be unreachable, since the
KafkaFuture#allOf already ensured
// that all of the futures completed successfully.
throw new RuntimeException(e);
}
});
return descriptions;
});
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
