You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2019/06/14 15:16:34 UTC
[kafka] branch trunk updated: KAFKA-8530;
Check for topic authorization errors in OffsetFetch response (#6928)
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 8dd4fb5 KAFKA-8530; Check for topic authorization errors in OffsetFetch response (#6928)
8dd4fb5 is described below
commit 8dd4fb5ebeccbb427ec5d0d92d4d5281ba9291f6
Author: Jason Gustafson <ja...@confluent.io>
AuthorDate: Fri Jun 14 08:16:11 2019 -0700
KAFKA-8530; Check for topic authorization errors in OffsetFetch response (#6928)
The OffsetFetch requires Topic Describe permission. If a client does not have this, we return TOPIC_AUTHORIZATION_FAILED at the partition level. Currently the consumer does not handle this error explicitly, but raises it as a generic `KafkaException`. For consistency with other APIs and to fix transient test failures in `PlaintextEndToEndAuthorizationTest`, we should raise `TopicAuthorizationFailedException` instead.
Reviewers: Ismael Juma <is...@juma.me.uk>
---
.../clients/consumer/internals/ConsumerCoordinator.java | 15 +++++++++++++--
.../kafka/common/requests/OffsetFetchResponse.java | 11 ++++++-----
.../consumer/internals/ConsumerCoordinatorTest.java | 16 ++++++++++++++++
3 files changed, 35 insertions(+), 7 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index bacb960..fbf5db4 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -959,6 +959,7 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
return;
}
+ Set<String> unauthorizedTopics = null;
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(response.responseData().size());
for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : response.responseData().entrySet()) {
TopicPartition tp = entry.getKey();
@@ -969,11 +970,17 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
future.raise(new KafkaException("Topic or Partition " + tp + " does not exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
} else {
future.raise(new KafkaException("Unexpected error in fetch offset response for partition " +
tp + ": " + error.message()));
+ return;
}
- return;
} else if (data.offset >= 0) {
// record the position with the offset (-1 indicates no committed offset to fetch)
offsets.put(tp, new OffsetAndMetadata(data.offset, data.leaderEpoch, data.metadata));
@@ -982,7 +989,11 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
}
}
- future.complete(offsets);
+ if (unauthorizedTopics != null) {
+ future.raise(new TopicAuthorizationException(unauthorizedTopics));
+ } else {
+ future.complete(offsets);
+ }
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java
index 2022bb7..ddf5c3b 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java
@@ -45,13 +45,14 @@ import static org.apache.kafka.common.protocol.CommonFields.TOPIC_NAME;
* Possible error codes:
*
* - Partition errors:
- * - UNKNOWN_TOPIC_OR_PARTITION (3)
+ * - {@link Errors#UNKNOWN_TOPIC_OR_PARTITION}
+ * - {@link Errors#TOPIC_AUTHORIZATION_FAILED}
*
* - Group or coordinator errors:
- * - COORDINATOR_LOAD_IN_PROGRESS (14)
- * - COORDINATOR_NOT_AVAILABLE (15)
- * - NOT_COORDINATOR (16)
- * - GROUP_AUTHORIZATION_FAILED (30)
+ * - {@link Errors#COORDINATOR_LOAD_IN_PROGRESS}
+ * - {@link Errors#COORDINATOR_NOT_AVAILABLE}
+ * - {@link Errors#NOT_COORDINATOR}
+ * - {@link Errors#GROUP_AUTHORIZATION_FAILED}
*/
public class OffsetFetchResponse extends AbstractResponse {
private static final Field.ComplexArray TOPICS = new Field.ComplexArray("responses",
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
index 9595639..4b377ea 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
@@ -35,6 +35,7 @@ import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.errors.FencedInstanceIdException;
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.apache.kafka.common.errors.OffsetMetadataTooLarge;
+import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.internals.ClusterResourceListeners;
import org.apache.kafka.common.internals.Topic;
@@ -1792,6 +1793,21 @@ public class ConsumerCoordinatorTest {
}
@Test
+ public void testTopicAuthorizationFailedInOffsetFetch() {
+ client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
+ coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));
+
+ OffsetFetchResponse.PartitionData data = new OffsetFetchResponse.PartitionData(-1, Optional.empty(),
+ "", Errors.TOPIC_AUTHORIZATION_FAILED);
+
+ client.prepareResponse(new OffsetFetchResponse(Errors.NONE, singletonMap(t1p, data)));
+ TopicAuthorizationException exception = assertThrows(TopicAuthorizationException.class, () ->
+ coordinator.fetchCommittedOffsets(singleton(t1p), time.timer(Long.MAX_VALUE)));
+
+ assertEquals(singleton(topic1), exception.unauthorizedTopics());
+ }
+
+ @Test
public void testRefreshOffsetLoadInProgress() {
client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));