You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/04/30 03:14:22 UTC
[GitHub] [kafka] hachikuji opened a new pull request #10616: KAFKA-12709; Add Admin API for `ListTransactions`
hachikuji opened a new pull request #10616:
URL: https://github.com/apache/kafka/pull/10616
This patch adds `Admin` support for the `listTransactions` API, which was added by [KIP-664](https://cwiki.apache.org/confluence/display/KAFKA/KIP-664%3A+Provide+tooling+to+detect+and+abort+hanging+transactions). Similar to `listConsumerGroups`, the new `listTransactions` API is intended to be sent to all brokers. This did not fit very well with the new `AdminApiDriver` that was used for the other KIP-664 APIs, so I had to refactor a bit. The way I ultimately handled it is to treat the brokerId as the key that needs to be mapped. Unlike with the other cases, the set of brokerIds is not known ahead of time and has to be discovered through a `Metadata` lookup (just as with `listConsumerGroups`). Because the future associated with `ListTransactionsResult` is more complex, I ended up externalizing future completion in a new `AdminApiFuture` object which is passed to `AdminApiDriver` during construction.
One minor change that is also worth calling out is the removal of `AdminApiHandler.Keys`. This was previously used to indicate the sets of statically and dynamically mapped keys. Instead, I now handle this with a new `StaticBrokerStrategy`. The static scope is indicated by returning an `ApiRequestScope` object which has a specific destination broker set.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10616: KAFKA-12709; Add Admin API for `ListTransactions`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10616:
URL: https://github.com/apache/kafka/pull/10616#discussion_r638764893
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategy.java
##########
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin.internals;
+
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.MetadataRequest;
+import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.OptionalInt;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * This class is used for use cases which require requests to be sent to all
+ * brokers in the cluster.
+ *
+ * This is a slightly degenerate case of a lookup strategy in the sense that
+ * the broker IDs are used as both the keys and values. Also, unlike
+ * {@link CoordinatorStrategy} and {@link PartitionLeaderStrategy}, we do not
+ * know the set of keys ahead of time: we require the initial lookup in order
+ * to discover what the broker IDs are. This is represented with a more complex
+ * type {@code Future<Map<Integer, Future<V>>} in the admin API result type.
+ * For example, see {@link org.apache.kafka.clients.admin.ListTransactionsResult}.
+ */
+public class AllBrokersStrategy implements AdminApiLookupStrategy<AllBrokersStrategy.BrokerKey> {
+ public static final BrokerKey ANY_BROKER = new BrokerKey(OptionalInt.empty());
+ public static final Set<BrokerKey> LOOKUP_KEYS = Collections.singleton(ANY_BROKER);
+ private static final ApiRequestScope SINGLE_REQUEST_SCOPE = new ApiRequestScope() {
+ };
+
+ private final Logger log;
+
+ public AllBrokersStrategy(
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(AllBrokersStrategy.class);
+ }
+
+ @Override
+ public ApiRequestScope lookupScope(BrokerKey key) {
+ return SINGLE_REQUEST_SCOPE;
+ }
+
+ @Override
+ public MetadataRequest.Builder buildRequest(Set<BrokerKey> keys) {
+ validateLookupKeys(keys);
+ // Send empty `Metadata` request. We are only interested in the brokers from the response
+ return new MetadataRequest.Builder(new MetadataRequestData());
+ }
+
+ @Override
+ public LookupResult<BrokerKey> handleResponse(Set<BrokerKey> keys, AbstractResponse abstractResponse) {
+ validateLookupKeys(keys);
+
+ MetadataResponse response = (MetadataResponse) abstractResponse;
+ MetadataResponseData.MetadataResponseBrokerCollection brokers = response.data().brokers();
+
+ if (brokers.isEmpty()) {
+ log.debug("Metadata response contained no brokers. Will backoff and retry");
+ return LookupResult.empty();
+ } else {
+ log.debug("Discovered all brokers {} to send requests to", brokers);
+ }
+
+ Map<BrokerKey, Integer> brokerKeys = brokers.stream().collect(Collectors.toMap(
+ broker -> new BrokerKey(OptionalInt.of(broker.nodeId())),
+ MetadataResponseData.MetadataResponseBroker::nodeId
+ ));
+
+ return new LookupResult<>(
+ Collections.singletonList(ANY_BROKER),
+ Collections.emptyMap(),
+ brokerKeys
+ );
+ }
+
+ private void validateLookupKeys(Set<BrokerKey> keys) {
+ if (keys.size() != 1) {
+ throw new IllegalArgumentException("Unexpected key set " + keys);
Review comment:
nit: I would add a `:` after `set` to stay consistent with the other error messages in this PR.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategy.java
##########
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin.internals;
+
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.MetadataRequest;
+import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.OptionalInt;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * This class is used for use cases which require requests to be sent to all
+ * brokers in the cluster.
+ *
+ * This is a slightly degenerate case of a lookup strategy in the sense that
+ * the broker IDs are used as both the keys and values. Also, unlike
+ * {@link CoordinatorStrategy} and {@link PartitionLeaderStrategy}, we do not
+ * know the set of keys ahead of time: we require the initial lookup in order
+ * to discover what the broker IDs are. This is represented with a more complex
+ * type {@code Future<Map<Integer, Future<V>>} in the admin API result type.
+ * For example, see {@link org.apache.kafka.clients.admin.ListTransactionsResult}.
+ */
+public class AllBrokersStrategy implements AdminApiLookupStrategy<AllBrokersStrategy.BrokerKey> {
+ public static final BrokerKey ANY_BROKER = new BrokerKey(OptionalInt.empty());
+ public static final Set<BrokerKey> LOOKUP_KEYS = Collections.singleton(ANY_BROKER);
+ private static final ApiRequestScope SINGLE_REQUEST_SCOPE = new ApiRequestScope() {
+ };
+
+ private final Logger log;
+
+ public AllBrokersStrategy(
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(AllBrokersStrategy.class);
+ }
+
+ @Override
+ public ApiRequestScope lookupScope(BrokerKey key) {
+ return SINGLE_REQUEST_SCOPE;
+ }
+
+ @Override
+ public MetadataRequest.Builder buildRequest(Set<BrokerKey> keys) {
+ validateLookupKeys(keys);
+ // Send empty `Metadata` request. We are only interested in the brokers from the response
+ return new MetadataRequest.Builder(new MetadataRequestData());
+ }
+
+ @Override
+ public LookupResult<BrokerKey> handleResponse(Set<BrokerKey> keys, AbstractResponse abstractResponse) {
+ validateLookupKeys(keys);
+
+ MetadataResponse response = (MetadataResponse) abstractResponse;
+ MetadataResponseData.MetadataResponseBrokerCollection brokers = response.data().brokers();
+
+ if (brokers.isEmpty()) {
+ log.debug("Metadata response contained no brokers. Will backoff and retry");
+ return LookupResult.empty();
+ } else {
+ log.debug("Discovered all brokers {} to send requests to", brokers);
+ }
+
+ Map<BrokerKey, Integer> brokerKeys = brokers.stream().collect(Collectors.toMap(
+ broker -> new BrokerKey(OptionalInt.of(broker.nodeId())),
+ MetadataResponseData.MetadataResponseBroker::nodeId
+ ));
+
+ return new LookupResult<>(
+ Collections.singletonList(ANY_BROKER),
+ Collections.emptyMap(),
+ brokerKeys
+ );
+ }
+
+ private void validateLookupKeys(Set<BrokerKey> keys) {
+ if (keys.size() != 1) {
+ throw new IllegalArgumentException("Unexpected key set " + keys);
+ }
+ BrokerKey key = keys.iterator().next();
+ if (key != ANY_BROKER) {
+ throw new IllegalArgumentException("Unexpected key set " + keys);
Review comment:
nit: ditto.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategy.java
##########
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin.internals;
+
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.MetadataRequest;
+import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.OptionalInt;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * This class is used for use cases which require requests to be sent to all
+ * brokers in the cluster.
+ *
+ * This is a slightly degenerate case of a lookup strategy in the sense that
+ * the broker IDs are used as both the keys and values. Also, unlike
+ * {@link CoordinatorStrategy} and {@link PartitionLeaderStrategy}, we do not
+ * know the set of keys ahead of time: we require the initial lookup in order
+ * to discover what the broker IDs are. This is represented with a more complex
+ * type {@code Future<Map<Integer, Future<V>>} in the admin API result type.
+ * For example, see {@link org.apache.kafka.clients.admin.ListTransactionsResult}.
+ */
+public class AllBrokersStrategy implements AdminApiLookupStrategy<AllBrokersStrategy.BrokerKey> {
+ public static final BrokerKey ANY_BROKER = new BrokerKey(OptionalInt.empty());
+ public static final Set<BrokerKey> LOOKUP_KEYS = Collections.singleton(ANY_BROKER);
+ private static final ApiRequestScope SINGLE_REQUEST_SCOPE = new ApiRequestScope() {
+ };
+
+ private final Logger log;
+
+ public AllBrokersStrategy(
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(AllBrokersStrategy.class);
+ }
+
+ @Override
+ public ApiRequestScope lookupScope(BrokerKey key) {
+ return SINGLE_REQUEST_SCOPE;
+ }
+
+ @Override
+ public MetadataRequest.Builder buildRequest(Set<BrokerKey> keys) {
+ validateLookupKeys(keys);
+ // Send empty `Metadata` request. We are only interested in the brokers from the response
+ return new MetadataRequest.Builder(new MetadataRequestData());
+ }
+
+ @Override
+ public LookupResult<BrokerKey> handleResponse(Set<BrokerKey> keys, AbstractResponse abstractResponse) {
+ validateLookupKeys(keys);
+
+ MetadataResponse response = (MetadataResponse) abstractResponse;
+ MetadataResponseData.MetadataResponseBrokerCollection brokers = response.data().brokers();
+
+ if (brokers.isEmpty()) {
+ log.debug("Metadata response contained no brokers. Will backoff and retry");
+ return LookupResult.empty();
+ } else {
+ log.debug("Discovered all brokers {} to send requests to", brokers);
+ }
+
+ Map<BrokerKey, Integer> brokerKeys = brokers.stream().collect(Collectors.toMap(
+ broker -> new BrokerKey(OptionalInt.of(broker.nodeId())),
+ MetadataResponseData.MetadataResponseBroker::nodeId
+ ));
+
+ return new LookupResult<>(
+ Collections.singletonList(ANY_BROKER),
+ Collections.emptyMap(),
+ brokerKeys
+ );
+ }
+
+ private void validateLookupKeys(Set<BrokerKey> keys) {
+ if (keys.size() != 1) {
+ throw new IllegalArgumentException("Unexpected key set " + keys);
+ }
+ BrokerKey key = keys.iterator().next();
+ if (key != ANY_BROKER) {
+ throw new IllegalArgumentException("Unexpected key set " + keys);
+ }
+ }
+
+ public static class BrokerKey {
+ public final OptionalInt brokerId;
+
+ public BrokerKey(OptionalInt brokerId) {
+ this.brokerId = brokerId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ BrokerKey that = (BrokerKey) o;
+ return Objects.equals(brokerId, that.brokerId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(brokerId);
+ }
+
+ @Override
+ public String toString() {
+ return "BrokerKey(" +
+ "brokerId=" + brokerId +
+ ')';
+ }
+ }
+
+ public static class AllBrokersFuture<V> implements AdminApiFuture<BrokerKey, V> {
+ private final KafkaFutureImpl<Map<Integer, KafkaFutureImpl<V>>> future = new KafkaFutureImpl<>();
+ private final Map<Integer, KafkaFutureImpl<V>> brokerFutures = new HashMap<>();
+
+ @Override
+ public Set<BrokerKey> lookupKeys() {
+ return LOOKUP_KEYS;
+ }
+
+ @Override
+ public void completeLookup(Map<BrokerKey, Integer> brokerMapping) {
+ brokerMapping.forEach((brokerKey, brokerId) -> {
+ if (brokerId != brokerKey.brokerId.orElse(-1)) {
+ throw new IllegalArgumentException("Invalid lookup mapping " + brokerKey + " -> " + brokerId);
+ }
+ brokerFutures.put(brokerId, new KafkaFutureImpl<>());
+ });
+ future.complete(brokerFutures);
+ }
+
+ @Override
+ public void completeLookupExceptionally(Map<BrokerKey, Throwable> lookupErrors) {
+ if (!LOOKUP_KEYS.equals(lookupErrors.keySet())) {
+ throw new IllegalArgumentException("Unexpected keys among lookup errors: " + lookupErrors);
+ }
+ future.completeExceptionally(lookupErrors.get(ANY_BROKER));
+ }
+
+ @Override
+ public void complete(Map<BrokerKey, V> values) {
+ values.forEach(this::complete);
+ }
+
+ private void complete(AllBrokersStrategy.BrokerKey key, V value) {
+ if (key == ANY_BROKER) {
+ throw new IllegalArgumentException("Invalid attempt to complete with lookup key sentinel");
+ } else {
+ futureOrThrow(key).complete(value);
+ }
+ }
+
+ @Override
+ public void completeExceptionally(Map<BrokerKey, Throwable> errors) {
+ errors.forEach(this::completeExceptionally);
+ }
+
+ private void completeExceptionally(AllBrokersStrategy.BrokerKey key, Throwable t) {
+ if (key == ANY_BROKER) {
+ future.completeExceptionally(t);
+ } else {
+ futureOrThrow(key).completeExceptionally(t);
+ }
+ }
+
+ public KafkaFutureImpl<Map<Integer, KafkaFutureImpl<V>>> all() {
+ return future;
+ }
+
+ private KafkaFutureImpl<V> futureOrThrow(BrokerKey key) {
+ if (!key.brokerId.isPresent()) {
+ throw new IllegalArgumentException("Attempt to complete with invalid key " + key);
Review comment:
nit: ditto.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategy.java
##########
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin.internals;
+
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.MetadataRequest;
+import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.OptionalInt;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * This class is used for use cases which require requests to be sent to all
+ * brokers in the cluster.
+ *
+ * This is a slightly degenerate case of a lookup strategy in the sense that
+ * the broker IDs are used as both the keys and values. Also, unlike
+ * {@link CoordinatorStrategy} and {@link PartitionLeaderStrategy}, we do not
+ * know the set of keys ahead of time: we require the initial lookup in order
+ * to discover what the broker IDs are. This is represented with a more complex
+ * type {@code Future<Map<Integer, Future<V>>} in the admin API result type.
+ * For example, see {@link org.apache.kafka.clients.admin.ListTransactionsResult}.
+ */
+public class AllBrokersStrategy implements AdminApiLookupStrategy<AllBrokersStrategy.BrokerKey> {
+ public static final BrokerKey ANY_BROKER = new BrokerKey(OptionalInt.empty());
+ public static final Set<BrokerKey> LOOKUP_KEYS = Collections.singleton(ANY_BROKER);
+ private static final ApiRequestScope SINGLE_REQUEST_SCOPE = new ApiRequestScope() {
+ };
+
+ private final Logger log;
+
+ public AllBrokersStrategy(
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(AllBrokersStrategy.class);
+ }
+
+ @Override
+ public ApiRequestScope lookupScope(BrokerKey key) {
+ return SINGLE_REQUEST_SCOPE;
+ }
+
+ @Override
+ public MetadataRequest.Builder buildRequest(Set<BrokerKey> keys) {
+ validateLookupKeys(keys);
+ // Send empty `Metadata` request. We are only interested in the brokers from the response
+ return new MetadataRequest.Builder(new MetadataRequestData());
+ }
+
+ @Override
+ public LookupResult<BrokerKey> handleResponse(Set<BrokerKey> keys, AbstractResponse abstractResponse) {
+ validateLookupKeys(keys);
+
+ MetadataResponse response = (MetadataResponse) abstractResponse;
+ MetadataResponseData.MetadataResponseBrokerCollection brokers = response.data().brokers();
+
+ if (brokers.isEmpty()) {
+ log.debug("Metadata response contained no brokers. Will backoff and retry");
+ return LookupResult.empty();
+ } else {
+ log.debug("Discovered all brokers {} to send requests to", brokers);
+ }
+
+ Map<BrokerKey, Integer> brokerKeys = brokers.stream().collect(Collectors.toMap(
+ broker -> new BrokerKey(OptionalInt.of(broker.nodeId())),
+ MetadataResponseData.MetadataResponseBroker::nodeId
+ ));
+
+ return new LookupResult<>(
+ Collections.singletonList(ANY_BROKER),
+ Collections.emptyMap(),
+ brokerKeys
+ );
+ }
+
+ private void validateLookupKeys(Set<BrokerKey> keys) {
+ if (keys.size() != 1) {
+ throw new IllegalArgumentException("Unexpected key set " + keys);
+ }
+ BrokerKey key = keys.iterator().next();
+ if (key != ANY_BROKER) {
+ throw new IllegalArgumentException("Unexpected key set " + keys);
+ }
+ }
+
+ public static class BrokerKey {
+ public final OptionalInt brokerId;
+
+ public BrokerKey(OptionalInt brokerId) {
+ this.brokerId = brokerId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ BrokerKey that = (BrokerKey) o;
+ return Objects.equals(brokerId, that.brokerId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(brokerId);
+ }
+
+ @Override
+ public String toString() {
+ return "BrokerKey(" +
+ "brokerId=" + brokerId +
+ ')';
+ }
+ }
+
+ public static class AllBrokersFuture<V> implements AdminApiFuture<BrokerKey, V> {
+ private final KafkaFutureImpl<Map<Integer, KafkaFutureImpl<V>>> future = new KafkaFutureImpl<>();
+ private final Map<Integer, KafkaFutureImpl<V>> brokerFutures = new HashMap<>();
+
+ @Override
+ public Set<BrokerKey> lookupKeys() {
+ return LOOKUP_KEYS;
+ }
+
+ @Override
+ public void completeLookup(Map<BrokerKey, Integer> brokerMapping) {
+ brokerMapping.forEach((brokerKey, brokerId) -> {
+ if (brokerId != brokerKey.brokerId.orElse(-1)) {
+ throw new IllegalArgumentException("Invalid lookup mapping " + brokerKey + " -> " + brokerId);
+ }
+ brokerFutures.put(brokerId, new KafkaFutureImpl<>());
+ });
+ future.complete(brokerFutures);
+ }
+
+ @Override
+ public void completeLookupExceptionally(Map<BrokerKey, Throwable> lookupErrors) {
+ if (!LOOKUP_KEYS.equals(lookupErrors.keySet())) {
+ throw new IllegalArgumentException("Unexpected keys among lookup errors: " + lookupErrors);
+ }
+ future.completeExceptionally(lookupErrors.get(ANY_BROKER));
+ }
+
+ @Override
+ public void complete(Map<BrokerKey, V> values) {
+ values.forEach(this::complete);
+ }
+
+ private void complete(AllBrokersStrategy.BrokerKey key, V value) {
+ if (key == ANY_BROKER) {
+ throw new IllegalArgumentException("Invalid attempt to complete with lookup key sentinel");
+ } else {
+ futureOrThrow(key).complete(value);
+ }
+ }
+
+ @Override
+ public void completeExceptionally(Map<BrokerKey, Throwable> errors) {
+ errors.forEach(this::completeExceptionally);
+ }
+
+ private void completeExceptionally(AllBrokersStrategy.BrokerKey key, Throwable t) {
+ if (key == ANY_BROKER) {
+ future.completeExceptionally(t);
+ } else {
+ futureOrThrow(key).completeExceptionally(t);
+ }
+ }
+
+ public KafkaFutureImpl<Map<Integer, KafkaFutureImpl<V>>> all() {
+ return future;
+ }
+
+ private KafkaFutureImpl<V> futureOrThrow(BrokerKey key) {
+ if (!key.brokerId.isPresent()) {
+ throw new IllegalArgumentException("Attempt to complete with invalid key " + key);
+ } else {
+ int brokerId = key.brokerId.getAsInt();
+ KafkaFutureImpl<V> future = brokerFutures.get(brokerId);
+ if (future == null) {
+ throw new IllegalArgumentException("Attempt to complete with unknown broker id " + brokerId);
Review comment:
nit: ditto.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsResult.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The result of the {@link Admin#listTransactions()} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListTransactionsResult {
+ private final KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future;
+
+ ListTransactionsResult(KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future) {
+ this.future = future;
+ }
+
+ public KafkaFuture<Collection<TransactionListing>> all() {
+ return allByBrokerId().thenApply(map -> {
+ List<TransactionListing> allListings = new ArrayList<>();
+ for (Collection<TransactionListing> listings : map.values()) {
+ allListings.addAll(listings);
+ }
+ return allListings;
+ });
+ }
+
+ public KafkaFuture<Set<Integer>> brokerIds() {
+ return future.thenApply(map -> new HashSet<>(map.keySet()));
+ }
Review comment:
This is the only remaining point of discussion. I don't have a strong preference for any of them so I leave it up to you.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #10616: KAFKA-12709; Add Admin API for `ListTransactions`
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #10616:
URL: https://github.com/apache/kafka/pull/10616
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10616: KAFKA-12709; Add Admin API for `ListTransactions`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10616:
URL: https://github.com/apache/kafka/pull/10616#discussion_r625165566
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsResult.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The result of the {@link Admin#listTransactions()} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListTransactionsResult {
+ private final KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future;
+
+ ListTransactionsResult(KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future) {
+ this.future = future;
+ }
+
+ public KafkaFuture<Collection<TransactionListing>> all() {
+ return allByBrokerId().thenApply(map -> {
+ List<TransactionListing> allListings = new ArrayList<>();
+ for (Collection<TransactionListing> listings : map.values()) {
+ allListings.addAll(listings);
+ }
+ return allListings;
+ });
+ }
+
+ public KafkaFuture<Set<Integer>> brokerIds() {
+ return future.thenApply(map -> new HashSet<>(map.keySet()));
+ }
Review comment:
I have a small preference for `Future<Map<Integer, Future>>` because it seems more aligned to how we do it for other APIs but I don't feel strong about it.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10616: KAFKA-12709; Add Admin API for `ListTransactions`
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10616:
URL: https://github.com/apache/kafka/pull/10616#discussion_r643496029
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsResult.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The result of the {@link Admin#listTransactions()} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListTransactionsResult {
+ private final KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future;
+
+ ListTransactionsResult(KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future) {
+ this.future = future;
+ }
+
+ public KafkaFuture<Collection<TransactionListing>> all() {
+ return allByBrokerId().thenApply(map -> {
+ List<TransactionListing> allListings = new ArrayList<>();
+ for (Collection<TransactionListing> listings : map.values()) {
+ allListings.addAll(listings);
+ }
+ return allListings;
+ });
+ }
+
+ public KafkaFuture<Set<Integer>> brokerIds() {
+ return future.thenApply(map -> new HashSet<>(map.keySet()));
+ }
Review comment:
I also don't feel too strong about it, so I decided to take the suggestion.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10616: KAFKA-12709; Add Admin API for `ListTransactions`
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10616:
URL: https://github.com/apache/kafka/pull/10616#discussion_r624062096
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsResult.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The result of the {@link Admin#listTransactions()} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListTransactionsResult {
+ private final KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future;
+
+ ListTransactionsResult(KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future) {
+ this.future = future;
+ }
+
+ public KafkaFuture<Collection<TransactionListing>> all() {
+ return allByBrokerId().thenApply(map -> {
+ List<TransactionListing> allListings = new ArrayList<>();
+ for (Collection<TransactionListing> listings : map.values()) {
+ allListings.addAll(listings);
+ }
+ return allListings;
+ });
+ }
+
+ public KafkaFuture<Set<Integer>> brokerIds() {
+ return future.thenApply(map -> new HashSet<>(map.keySet()));
+ }
Review comment:
It was to make `byBrokerId` useful. If a user wants to be able to collect partial results, then they can do something like this:
```java
Set<Integer> brokerIds = result.brokerIds().get();
for (Integer brokerId : brokerIds) {
KafkaFuture<Collection<TransactionListing>> future = result.byBrokerId(brokerId);
try {
Collection<TransactionListing> listings =future.get();
...
} catch (Exception e) {
...
}
}
```
I think an alternative would be to expose the underlying `Future<Map<Integer, Future>>`. What do you think?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10616: KAFKA-12709; Add Admin API for `ListTransactions`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10616:
URL: https://github.com/apache/kafka/pull/10616#discussion_r623846200
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsOptions.java
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Options for {@link Admin#listTransactions()}.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListTransactionsOptions extends AbstractOptions<ListTransactionsOptions> {
+ private Set<TransactionState> filteredStates = Collections.emptySet();
+ private Set<Long> filteredProducerIds = Collections.emptySet();
+
+ public ListTransactionsOptions filterStates(Collection<TransactionState> states) {
Review comment:
Should we add javadoc to the various methods here?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsResult.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The result of the {@link Admin#listTransactions()} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListTransactionsResult {
+ private final KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future;
+
+ ListTransactionsResult(KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future) {
+ this.future = future;
+ }
+
+ public KafkaFuture<Collection<TransactionListing>> all() {
+ return allByBrokerId().thenApply(map -> {
+ List<TransactionListing> allListings = new ArrayList<>();
+ for (Collection<TransactionListing> listings : map.values()) {
+ allListings.addAll(listings);
+ }
+ return allListings;
+ });
+ }
+
+ public KafkaFuture<Set<Integer>> brokerIds() {
+ return future.thenApply(map -> new HashSet<>(map.keySet()));
+ }
Review comment:
I wonder if this one is really necessary. What's the reasoning behind it?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListTransactionsOptions;
+import org.apache.kafka.clients.admin.TransactionListing;
+import org.apache.kafka.clients.admin.TransactionState;
+import org.apache.kafka.common.errors.CoordinatorNotAvailableException;
+import org.apache.kafka.common.message.ListTransactionsRequestData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListTransactionsRequest;
+import org.apache.kafka.common.requests.ListTransactionsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListTransactionsHandler implements AdminApiHandler<AllBrokersStrategy.BrokerKey, Collection<TransactionListing>> {
+ private final Logger log;
+ private final ListTransactionsOptions options;
+ private final AllBrokersStrategy lookupStrategy;
+
+ public ListTransactionsHandler(
+ ListTransactionsOptions options,
+ LogContext logContext
+ ) {
+ this.options = options;
+ this.log = logContext.logger(ListTransactionsHandler.class);
+ this.lookupStrategy = new AllBrokersStrategy(logContext);
+ }
+
+ public static AllBrokersStrategy.AllBrokersFuture<Collection<TransactionListing>> newFuture() {
+ return new AllBrokersStrategy.AllBrokersFuture<>();
+ }
+
+ @Override
+ public String apiName() {
+ return "listTransactions";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<AllBrokersStrategy.BrokerKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ @Override
+ public ListTransactionsRequest.Builder buildRequest(
+ int brokerId,
+ Set<AllBrokersStrategy.BrokerKey> keys
+ ) {
+ ListTransactionsRequestData request = new ListTransactionsRequestData();
+ request.setProducerIdFilters(new ArrayList<>(options.filteredProducerIds()));
+ request.setStateFilters(options.filteredStates().stream()
+ .map(TransactionState::toString)
+ .collect(Collectors.toList()));
+ return new ListTransactionsRequest.Builder(request);
+ }
+
+ @Override
+ public ApiResult<AllBrokersStrategy.BrokerKey, Collection<TransactionListing>> handleResponse(
+ int brokerId,
+ Set<AllBrokersStrategy.BrokerKey> keys,
+ AbstractResponse abstractResponse
+ ) {
+ AllBrokersStrategy.BrokerKey key = requireSingleton(keys, brokerId);
+
+ ListTransactionsResponse response = (ListTransactionsResponse) abstractResponse;
+ Errors error = Errors.forCode(response.data().errorCode());
+
+ if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ log.debug("The `ListTransactions` request sent to broker {} failed because the " +
+ "coordinator is still loading state. Will try again after backing off", brokerId);
+ return ApiResult.empty();
+ } else if (error == Errors.COORDINATOR_NOT_AVAILABLE) {
+ log.debug("The `ListTransactions` request sent to broker {} failed because the " +
+ "coordinator is shutting down", brokerId);
+ return ApiResult.failed(key, new CoordinatorNotAvailableException("ListTransactions " +
+ "request sent to broker " + brokerId + " failed because the coordinator is shutting down"));
+ } else if (error != Errors.NONE) {
+ log.error("The `ListTransactions` request sent to broker {} failed because of an " +
+ "unexpected error {}", brokerId, error);
+ return ApiResult.failed(key, error.exception("ListTransactions request " +
+ "sent to broker " + brokerId + " failed with an unexpected exception"));
+ } else {
+ List<TransactionListing> listings = response.data().transactionStates().stream()
+ .map(transactionState -> new TransactionListing(
+ transactionState.transactionalId(),
+ transactionState.producerId(),
+ TransactionState.parse(transactionState.transactionState())))
+ .collect(Collectors.toList());
+ return ApiResult.completed(key, listings);
+ }
+ }
+
+ private AllBrokersStrategy.BrokerKey requireSingleton(
+ Set<AllBrokersStrategy.BrokerKey> keys,
+ int brokerId
+ ) {
+ if (keys.size() != 1) {
+ throw new IllegalArgumentException("Unexpected key set" + keys);
Review comment:
nit: `Unexpected key set` -> `Unexpected key set: `. The `:` might not be necessary but we should add a space. The same applies to the `IllegalArgumentException` below.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsResult.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The result of the {@link Admin#listTransactions()} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListTransactionsResult {
+ private final KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future;
+
+ ListTransactionsResult(KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future) {
+ this.future = future;
+ }
+
+ public KafkaFuture<Collection<TransactionListing>> all() {
+ return allByBrokerId().thenApply(map -> {
+ List<TransactionListing> allListings = new ArrayList<>();
+ for (Collection<TransactionListing> listings : map.values()) {
+ allListings.addAll(listings);
+ }
+ return allListings;
+ });
+ }
+
+ public KafkaFuture<Set<Integer>> brokerIds() {
+ return future.thenApply(map -> new HashSet<>(map.keySet()));
+ }
+
+ public KafkaFuture<Map<Integer, Collection<TransactionListing>>> allByBrokerId() {
+ KafkaFutureImpl<Map<Integer, Collection<TransactionListing>>> allFuture = new KafkaFutureImpl<>();
+ Map<Integer, Collection<TransactionListing>> allListingsMap = new HashMap<>();
+
+ future.whenComplete((map, topLevelException) -> {
+ if (topLevelException != null) {
+ allFuture.completeExceptionally(topLevelException);
+ return;
+ }
+
+ Set<Integer> remainingResponses = new HashSet<>(map.keySet());
+ for (Map.Entry<Integer, KafkaFutureImpl<Collection<TransactionListing>>> entry : map.entrySet()) {
+ Integer brokerId = entry.getKey();
+ KafkaFutureImpl<Collection<TransactionListing>> future = entry.getValue();
+ future.whenComplete((listings, brokerException) -> {
+
Review comment:
nit: This empty line is not necessary.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
##########
@@ -1533,6 +1533,28 @@ default AbortTransactionResult abortTransaction(AbortTransactionSpec spec) {
*/
AbortTransactionResult abortTransaction(AbortTransactionSpec spec, AbortTransactionOptions options);
+ /**
+ * List active transactions in the cluster. See
+ * {@link #listTransactions(ListTransactionsOptions)} for more details.
+ *
+ * @return The result
+ */
+ default ListTransactionsResult listTransactions() {
+ return listTransactions(new ListTransactionsOptions());
+ }
+
+ /**
+ * List active transactions in the cluster. This will query all potential transaction
+ * coordinators in the cluster and collect the state of all transactionalIds. Users
Review comment:
nit: `transactionalIds` -> `transactions`?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsResult.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The result of the {@link Admin#listTransactions()} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListTransactionsResult {
+ private final KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future;
+
+ ListTransactionsResult(KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future) {
+ this.future = future;
+ }
+
+ public KafkaFuture<Collection<TransactionListing>> all() {
+ return allByBrokerId().thenApply(map -> {
+ List<TransactionListing> allListings = new ArrayList<>();
+ for (Collection<TransactionListing> listings : map.values()) {
+ allListings.addAll(listings);
+ }
+ return allListings;
+ });
+ }
+
+ public KafkaFuture<Set<Integer>> brokerIds() {
+ return future.thenApply(map -> new HashSet<>(map.keySet()));
+ }
+
+ public KafkaFuture<Map<Integer, Collection<TransactionListing>>> allByBrokerId() {
+ KafkaFutureImpl<Map<Integer, Collection<TransactionListing>>> allFuture = new KafkaFutureImpl<>();
+ Map<Integer, Collection<TransactionListing>> allListingsMap = new HashMap<>();
+
+ future.whenComplete((map, topLevelException) -> {
+ if (topLevelException != null) {
+ allFuture.completeExceptionally(topLevelException);
+ return;
+ }
+
+ Set<Integer> remainingResponses = new HashSet<>(map.keySet());
+ for (Map.Entry<Integer, KafkaFutureImpl<Collection<TransactionListing>>> entry : map.entrySet()) {
+ Integer brokerId = entry.getKey();
+ KafkaFutureImpl<Collection<TransactionListing>> future = entry.getValue();
Review comment:
nit: We could use `forEach` here. It makes the code more concise.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org