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 2017/01/17 18:42:18 UTC
kafka git commit: MINOR: Some cleanups and additional testing for
KIP-88
Repository: kafka
Updated Branches:
refs/heads/trunk 55abe65e0 -> 7a84b241e
MINOR: Some cleanups and additional testing for KIP-88
Author: Jason Gustafson <ja...@confluent.io>
Reviewers: Vahid Hashemian <va...@us.ibm.com>, Ismael Juma <is...@juma.me.uk>
Closes #2383 from hachikuji/minor-cleanup-kip-88
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/7a84b241
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/7a84b241
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/7a84b241
Branch: refs/heads/trunk
Commit: 7a84b241eeb8cb63400a9512b066c3f733f94b8c
Parents: 55abe65
Author: Jason Gustafson <ja...@confluent.io>
Authored: Tue Jan 17 10:42:05 2017 -0800
Committer: Jason Gustafson <ja...@confluent.io>
Committed: Tue Jan 17 10:42:05 2017 -0800
----------------------------------------------------------------------
.../common/requests/OffsetFetchRequest.java | 24 +++++-
.../common/requests/OffsetFetchResponse.java | 83 ++++++-------------
.../clients/consumer/KafkaConsumerTest.java | 2 +-
.../internals/ConsumerCoordinatorTest.java | 4 +-
.../common/requests/RequestResponseTest.java | 23 ++++--
.../scala/kafka/api/OffsetFetchRequest.scala | 1 -
.../kafka/coordinator/GroupCoordinator.scala | 4 +-
.../coordinator/GroupMetadataManager.scala | 31 ++++---
.../src/main/scala/kafka/server/KafkaApis.scala | 86 ++++++++------------
.../kafka/api/AuthorizerIntegrationTest.scala | 37 ++++++++-
.../GroupCoordinatorResponseTest.scala | 58 ++++++++++++-
11 files changed, 211 insertions(+), 142 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a84b241/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
index 0ff49be..553fd96 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
@@ -23,6 +23,7 @@ import org.apache.kafka.common.utils.Utils;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -78,7 +79,7 @@ public class OffsetFetchRequest extends AbstractRequest {
private final List<TopicPartition> partitions;
public static OffsetFetchRequest forAllPartitions(String groupId) {
- return new OffsetFetchRequest.Builder(groupId, (List<TopicPartition>) null).setVersion((short) 2).build();
+ return new OffsetFetchRequest.Builder(groupId, null).setVersion((short) 2).build();
}
// v0, v1, and v2 have the same fields.
@@ -131,20 +132,35 @@ public class OffsetFetchRequest extends AbstractRequest {
groupId = struct.getString(GROUP_ID_KEY_NAME);
}
- @Override
- public AbstractResponse getErrorResponse(Throwable e) {
+ public OffsetFetchResponse getErrorResponse(Errors error) {
short versionId = version();
+
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responsePartitions = new HashMap<>();
+ if (versionId < 2) {
+ for (TopicPartition partition : this.partitions) {
+ responsePartitions.put(partition, new OffsetFetchResponse.PartitionData(
+ OffsetFetchResponse.INVALID_OFFSET,
+ OffsetFetchResponse.NO_METADATA,
+ error));
+ }
+ }
+
switch (versionId) {
case 0:
case 1:
case 2:
- return new OffsetFetchResponse(Errors.forException(e), partitions, versionId);
+ return new OffsetFetchResponse(error, responsePartitions, versionId);
default:
throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d",
versionId, this.getClass().getSimpleName(), ProtoUtils.latestVersion(ApiKeys.OFFSET_FETCH.id)));
}
}
+ @Override
+ public OffsetFetchResponse getErrorResponse(Throwable e) {
+ return getErrorResponse(Errors.forException(e));
+ }
+
public String groupId() {
return groupId;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a84b241/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java
----------------------------------------------------------------------
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 0095f38..9c14155 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
@@ -46,6 +46,8 @@ public class OffsetFetchResponse extends AbstractResponse {
public static final long INVALID_OFFSET = -1L;
public static final String NO_METADATA = "";
+ public static final PartitionData UNKNOWN_PARTITION = new PartitionData(INVALID_OFFSET, NO_METADATA,
+ Errors.UNKNOWN_TOPIC_OR_PARTITION);
/**
* Possible error codes:
@@ -59,7 +61,7 @@ public class OffsetFetchResponse extends AbstractResponse {
* - GROUP_AUTHORIZATION_FAILED (30)
*/
- public static final List<Errors> PARTITION_ERRORS = Arrays.asList(
+ private static final List<Errors> PARTITION_ERRORS = Arrays.asList(
Errors.UNKNOWN_TOPIC_OR_PARTITION,
Errors.TOPIC_AUTHORIZATION_FAILED);
@@ -82,14 +84,30 @@ public class OffsetFetchResponse extends AbstractResponse {
}
}
- private List<Struct> getTopicArray(Map<TopicPartition, PartitionData> responseData) {
- Map<String, Map<Integer, PartitionData>> topicsData = CollectionUtils.groupDataByTopic(responseData);
+ /**
+ * Constructor for the latest version.
+ * @param error Potential coordinator or group level error code
+ * @param responseData Fetched offset information grouped by topic-partition
+ */
+ public OffsetFetchResponse(Errors error, Map<TopicPartition, PartitionData> responseData) {
+ this(error, responseData, CURRENT_VERSION);
+ }
- List<Struct> topicArray = new ArrayList<Struct>();
+ /**
+ * Unified constructor for all versions.
+ * @param error Potential coordinator or group level error code (for api version 2 and later)
+ * @param responseData Fetched offset information grouped by topic-partition
+ * @param version The request API version
+ */
+ public OffsetFetchResponse(Errors error, Map<TopicPartition, PartitionData> responseData, int version) {
+ super(new Struct(ProtoUtils.responseSchema(ApiKeys.OFFSET_FETCH.id, version)));
+
+ Map<String, Map<Integer, PartitionData>> topicsData = CollectionUtils.groupDataByTopic(responseData);
+ List<Struct> topicArray = new ArrayList<>();
for (Map.Entry<String, Map<Integer, PartitionData>> entries : topicsData.entrySet()) {
Struct topicData = this.struct.instance(RESPONSES_KEY_NAME);
topicData.set(TOPIC_KEY_NAME, entries.getKey());
- List<Struct> partitionArray = new ArrayList<Struct>();
+ List<Struct> partitionArray = new ArrayList<>();
for (Map.Entry<Integer, PartitionData> partitionEntry : entries.getValue().entrySet()) {
PartitionData fetchPartitionData = partitionEntry.getValue();
Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME);
@@ -103,66 +121,17 @@ public class OffsetFetchResponse extends AbstractResponse {
topicArray.add(topicData);
}
- return topicArray;
- }
-
- /**
- * Unified constructor
- * @param responseData Fetched offset information grouped by topic-partition
- * @param topLevelErrorCode Potential coordinator or group level error code (for api version 2 and later)
- * @param version The request API version
- */
- public OffsetFetchResponse(Errors topLevelError, Map<TopicPartition, PartitionData> responseData, int version) {
- super(new Struct(ProtoUtils.responseSchema(ApiKeys.OFFSET_FETCH.id, version)));
-
- this.struct.set(RESPONSES_KEY_NAME, getTopicArray(responseData).toArray());
+ this.struct.set(RESPONSES_KEY_NAME, topicArray.toArray());
this.responseData = responseData;
- this.error = topLevelError;
+ this.error = error;
if (version > 1)
this.struct.set(ERROR_CODE_KEY_NAME, this.error.code());
}
- /**
- * Unified constructor (used only if there are errors in the response)
- * @param partitions partitions to be included in the response
- * @param topLevelErrorCode The error code to be reported in the response
- * @param version The request API version
- */
- public OffsetFetchResponse(Errors topLevelError, List<TopicPartition> partitions, int version) {
- super(new Struct(ProtoUtils.responseSchema(ApiKeys.OFFSET_FETCH.id, version)));
-
- assert topLevelError != Errors.NONE;
- this.responseData = new HashMap<>();
- this.error = topLevelError;
- if (version < 2) {
- for (TopicPartition partition : partitions) {
- this.responseData.put(partition, new OffsetFetchResponse.PartitionData(
- OffsetFetchResponse.INVALID_OFFSET,
- OffsetFetchResponse.NO_METADATA,
- topLevelError));
- }
- } else
- this.struct.set(ERROR_CODE_KEY_NAME, this.error.code());
-
- this.struct.set(RESPONSES_KEY_NAME, getTopicArray(this.responseData).toArray());
- }
-
- public OffsetFetchResponse(Map<TopicPartition, PartitionData> responseData) {
- this(Errors.NONE, responseData, CURRENT_VERSION);
- }
-
- /**
- * Constructor for version 2 and above when there is a coordinator or group level error
- * @param topLevelErrorCode Coordinator or group level error code
- */
- public OffsetFetchResponse(Errors topLevelError) {
- this(topLevelError, new ArrayList<TopicPartition>(), CURRENT_VERSION);
- }
-
public OffsetFetchResponse(Struct struct) {
super(struct);
Errors topLevelError = Errors.NONE;
- this.responseData = new HashMap<TopicPartition, PartitionData>();
+ this.responseData = new HashMap<>();
for (Object topicResponseObj : struct.getArray(RESPONSES_KEY_NAME)) {
Struct topicResponse = (Struct) topicResponseObj;
String topic = topicResponse.getString(TOPIC_KEY_NAME);
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a84b241/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
index 2eeed55..d4913df 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
@@ -1370,7 +1370,7 @@ public class KafkaConsumerTest {
for (Map.Entry<TopicPartition, Long> entry : offsets.entrySet()) {
partitionData.put(entry.getKey(), new OffsetFetchResponse.PartitionData(entry.getValue(), "", error));
}
- return new OffsetFetchResponse(partitionData);
+ return new OffsetFetchResponse(Errors.NONE, partitionData);
}
private ListOffsetResponse listOffsetsResponse(Map<TopicPartition, Long> offsets, short error) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a84b241/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
----------------------------------------------------------------------
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 9a8c0b9..ee6afe1 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
@@ -1482,12 +1482,12 @@ public class ConsumerCoordinatorTest {
}
private OffsetFetchResponse offsetFetchResponse(Errors topLevelError) {
- return new OffsetFetchResponse(topLevelError);
+ return new OffsetFetchResponse(topLevelError, Collections.<TopicPartition, OffsetFetchResponse.PartitionData>emptyMap());
}
private OffsetFetchResponse offsetFetchResponse(TopicPartition tp, Errors partitionLevelError, String metadata, long offset) {
OffsetFetchResponse.PartitionData data = new OffsetFetchResponse.PartitionData(offset, metadata, partitionLevelError);
- return new OffsetFetchResponse(Collections.singletonMap(tp, data));
+ return new OffsetFetchResponse(Errors.NONE, Collections.singletonMap(tp, data));
}
private OffsetCommitCallback callback(final AtomicBoolean success) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a84b241/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
index 0d3a1a8..a5ed806 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
@@ -43,6 +43,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import static java.util.Collections.singletonList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -90,8 +91,13 @@ public class RequestResponseTest {
checkSerialization(createOffsetCommitResponse(), null);
checkSerialization(OffsetFetchRequest.forAllPartitions("group1"));
checkSerialization(OffsetFetchRequest.forAllPartitions("group1").getErrorResponse(new NotCoordinatorForGroupException()), 2);
- checkSerialization(createOffsetFetchRequest());
- checkSerialization(createOffsetFetchRequest().getErrorResponse(new UnknownServerException()), null);
+ checkSerialization(createOffsetFetchRequest(0));
+ checkSerialization(createOffsetFetchRequest(1));
+ checkSerialization(createOffsetFetchRequest(2));
+ checkSerialization(OffsetFetchRequest.forAllPartitions("group1"));
+ checkSerialization(createOffsetFetchRequest(0).getErrorResponse(new UnknownServerException()), 0);
+ checkSerialization(createOffsetFetchRequest(1).getErrorResponse(new UnknownServerException()), 1);
+ checkSerialization(createOffsetFetchRequest(2).getErrorResponse(new UnknownServerException()), 2);
checkSerialization(createOffsetFetchResponse(), null);
checkSerialization(createProduceRequest());
checkSerialization(createProduceRequest().getErrorResponse(new UnknownServerException()), null);
@@ -337,7 +343,7 @@ public class RequestResponseTest {
}
private DescribeGroupsRequest createDescribeGroupRequest() {
- return new DescribeGroupsRequest.Builder(Collections.singletonList("test-group")).build();
+ return new DescribeGroupsRequest.Builder(singletonList("test-group")).build();
}
private DescribeGroupsResponse createDescribeGroupResponse() {
@@ -428,16 +434,17 @@ public class RequestResponseTest {
return new OffsetCommitResponse(responseData);
}
- private OffsetFetchRequest createOffsetFetchRequest() {
- return new OffsetFetchRequest.Builder("group1",
- Arrays.asList(new TopicPartition("test11", 1))).build();
+ private OffsetFetchRequest createOffsetFetchRequest(int version) {
+ return new OffsetFetchRequest.Builder("group1", singletonList(new TopicPartition("test11", 1)))
+ .setVersion((short) version)
+ .build();
}
private OffsetFetchResponse createOffsetFetchResponse() {
Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData = new HashMap<>();
responseData.put(new TopicPartition("test", 0), new OffsetFetchResponse.PartitionData(100L, "", Errors.NONE));
responseData.put(new TopicPartition("test", 1), new OffsetFetchResponse.PartitionData(100L, null, Errors.NONE));
- return new OffsetFetchResponse(responseData);
+ return new OffsetFetchResponse(Errors.NONE, responseData);
}
private ProduceRequest createProduceRequest() {
@@ -544,7 +551,7 @@ public class RequestResponseTest {
}
private SaslHandshakeResponse createSaslHandshakeResponse() {
- return new SaslHandshakeResponse(Errors.NONE.code(), Collections.singletonList("GSSAPI"));
+ return new SaslHandshakeResponse(Errors.NONE.code(), singletonList("GSSAPI"));
}
private ApiVersionsRequest createApiVersionRequest() {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a84b241/core/src/main/scala/kafka/api/OffsetFetchRequest.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/OffsetFetchRequest.scala b/core/src/main/scala/kafka/api/OffsetFetchRequest.scala
index 2908901..dac4cc5 100644
--- a/core/src/main/scala/kafka/api/OffsetFetchRequest.scala
+++ b/core/src/main/scala/kafka/api/OffsetFetchRequest.scala
@@ -25,7 +25,6 @@ import kafka.network.{RequestOrResponseSend, RequestChannel}
import kafka.network.RequestChannel.Response
import kafka.utils.Logging
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
-import org.apache.kafka.common.requests.OffsetFetchResponse.PARTITION_ERRORS
object OffsetFetchRequest extends Logging {
val CurrentVersion: Short = 2
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a84b241/core/src/main/scala/kafka/coordinator/GroupCoordinator.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/coordinator/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/GroupCoordinator.scala
index 4cbfad6..7abbc6e 100644
--- a/core/src/main/scala/kafka/coordinator/GroupCoordinator.scala
+++ b/core/src/main/scala/kafka/coordinator/GroupCoordinator.scala
@@ -422,7 +422,7 @@ class GroupCoordinator(val brokerId: Int,
}
}
- def doCommitOffsets(group: GroupMetadata,
+ private def doCommitOffsets(group: GroupMetadata,
memberId: String,
generationId: Int,
offsetMetadata: immutable.Map[TopicPartition, OffsetAndMetadata],
@@ -455,7 +455,7 @@ class GroupCoordinator(val brokerId: Int,
}
def handleFetchOffsets(groupId: String,
- partitions: Option[Seq[TopicPartition]]): (Errors, Map[TopicPartition, OffsetFetchResponse.PartitionData]) = {
+ partitions: Option[Seq[TopicPartition]] = None): (Errors, Map[TopicPartition, OffsetFetchResponse.PartitionData]) = {
if (!isActive.get)
(Errors.GROUP_COORDINATOR_NOT_AVAILABLE, Map())
else if (!isCoordinatorForGroup(groupId)) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a84b241/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala
index 74b46ad..2d6889c 100644
--- a/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala
+++ b/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala
@@ -97,7 +97,7 @@ class GroupMetadataManager(val brokerId: Int,
unit = TimeUnit.MILLISECONDS)
}
- def currentGroups(): Iterable[GroupMetadata] = groupMetadataCache.values
+ def currentGroups: Iterable[GroupMetadata] = groupMetadataCache.values
def isPartitionOwned(partition: Int) = inLock(partitionLock) { ownedPartitions.contains(partition) }
@@ -342,19 +342,24 @@ class GroupMetadataManager(val brokerId: Int,
(topicPartition, new OffsetFetchResponse.PartitionData(OffsetFetchResponse.INVALID_OFFSET, "", Errors.NONE))
}.toMap
} else {
- if (topicPartitionsOpt.isEmpty) {
- // Return offsets for all partitions owned by this consumer group. (this only applies to consumers that commit offsets to Kafka.)
- group.allOffsets.map { case (topicPartition, offsetAndMetadata) =>
- (topicPartition, new OffsetFetchResponse.PartitionData(offsetAndMetadata.offset, offsetAndMetadata.metadata, Errors.NONE))
- }
- } else {
- topicPartitionsOpt.getOrElse(Seq.empty[TopicPartition]).map { topicPartition =>
- group.offset(topicPartition) match {
- case None => (topicPartition, new OffsetFetchResponse.PartitionData(OffsetFetchResponse.INVALID_OFFSET, "", Errors.NONE))
- case Some(offsetAndMetadata) =>
- (topicPartition, new OffsetFetchResponse.PartitionData(offsetAndMetadata.offset, offsetAndMetadata.metadata, Errors.NONE))
+ topicPartitionsOpt match {
+ case None =>
+ // Return offsets for all partitions owned by this consumer group. (this only applies to consumers
+ // that commit offsets to Kafka.)
+ group.allOffsets.map { case (topicPartition, offsetAndMetadata) =>
+ topicPartition -> new OffsetFetchResponse.PartitionData(offsetAndMetadata.offset, offsetAndMetadata.metadata, Errors.NONE)
}
- }.toMap
+
+ case Some(topicPartitions) =>
+ topicPartitionsOpt.getOrElse(Seq.empty[TopicPartition]).map { topicPartition =>
+ val partitionData = group.offset(topicPartition) match {
+ case None =>
+ new OffsetFetchResponse.PartitionData(OffsetFetchResponse.INVALID_OFFSET, "", Errors.NONE)
+ case Some(offsetAndMetadata) =>
+ new OffsetFetchResponse.PartitionData(offsetAndMetadata.offset, offsetAndMetadata.metadata, Errors.NONE)
+ }
+ topicPartition -> partitionData
+ }.toMap
}
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a84b241/core/src/main/scala/kafka/server/KafkaApis.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index a6ad7b2..530dafc 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -38,7 +38,7 @@ import kafka.utils.{Logging, ZKGroupTopicDirs, ZkUtils}
import org.apache.kafka.common.errors.{ClusterAuthorizationException, NotLeaderForPartitionException, TopicExistsException, UnknownTopicOrPartitionException, UnsupportedForMessageFormatException}
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.network.ListenerName
-import org.apache.kafka.common.protocol.{ApiKeys, Errors, Protocol, SecurityProtocol}
+import org.apache.kafka.common.protocol.{ApiKeys, Errors, Protocol}
import org.apache.kafka.common.record.{MemoryRecords, Record}
import org.apache.kafka.common.requests._
import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
@@ -886,38 +886,31 @@ class KafkaApis(val requestChannel: RequestChannel,
requestChannel.sendResponse(new RequestChannel.Response(request, responseBody))
}
- /*
+ /**
* Handle an offset fetch request
*/
def handleOffsetFetchRequest(request: RequestChannel.Request) {
val header = request.header
val offsetFetchRequest = request.body.asInstanceOf[OffsetFetchRequest]
+ def authorizeTopicDescribe(partition: TopicPartition) =
+ authorize(request.session, Describe, new Resource(auth.Topic, partition.topic))
+
val offsetFetchResponse =
// reject the request if not authorized to the group
if (!authorize(request.session, Read, new Resource(Group, offsetFetchRequest.groupId)))
- new OffsetFetchResponse(Errors.GROUP_AUTHORIZATION_FAILED, offsetFetchRequest.partitions, header.apiVersion)
+ offsetFetchRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED)
else {
- val partitions =
- if (offsetFetchRequest.isAllPartitions)
- List[TopicPartition]()
- else
- offsetFetchRequest.partitions.asScala.toList
-
- val (authorizedPartitions, unauthorizedPartitions) =
- partitions.partition { partition => authorize(request.session, Describe, new Resource(auth.Topic, partition.topic)) }
-
- val unknownTopicPartitionResponse = new OffsetFetchResponse.PartitionData(
- OffsetFetchResponse.INVALID_OFFSET, OffsetFetchResponse.NO_METADATA, Errors.UNKNOWN_TOPIC_OR_PARTITION)
- val unauthorizedStatus = unauthorizedPartitions.map(topicPartition => (topicPartition, unknownTopicPartitionResponse)).toMap
-
if (header.apiVersion == 0) {
+ val (authorizedPartitions, unauthorizedPartitions) = offsetFetchRequest.partitions.asScala
+ .partition(authorizeTopicDescribe)
+
// version 0 reads offsets from ZK
- val responseInfo = authorizedPartitions.map { topicPartition =>
+ val authorizedPartitionData = authorizedPartitions.map { topicPartition =>
val topicDirs = new ZKGroupTopicDirs(offsetFetchRequest.groupId, topicPartition.topic)
try {
if (!metadataCache.contains(topicPartition.topic))
- (topicPartition, unknownTopicPartitionResponse)
+ (topicPartition, OffsetFetchResponse.UNKNOWN_PARTITION)
else {
val payloadOpt = zkUtils.readDataMaybeNull(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}")._1
payloadOpt match {
@@ -925,7 +918,7 @@ class KafkaApis(val requestChannel: RequestChannel,
(topicPartition, new OffsetFetchResponse.PartitionData(
payload.toLong, OffsetFetchResponse.NO_METADATA, Errors.NONE))
case None =>
- (topicPartition, unknownTopicPartitionResponse)
+ (topicPartition, OffsetFetchResponse.UNKNOWN_PARTITION)
}
}
} catch {
@@ -934,43 +927,32 @@ class KafkaApis(val requestChannel: RequestChannel,
OffsetFetchResponse.INVALID_OFFSET, OffsetFetchResponse.NO_METADATA, Errors.forException(e)))
}
}.toMap
- new OffsetFetchResponse(Errors.NONE, (responseInfo ++ unauthorizedStatus).asJava, header.apiVersion)
- }
- else {
+
+ val unauthorizedPartitionData = unauthorizedPartitions.map(_ -> OffsetFetchResponse.UNKNOWN_PARTITION).toMap
+ new OffsetFetchResponse(Errors.NONE, (authorizedPartitionData ++ unauthorizedPartitionData).asJava, header.apiVersion)
+ } else {
// versions 1 and above read offsets from Kafka
- val offsets = coordinator.handleFetchOffsets(offsetFetchRequest.groupId,
- if (offsetFetchRequest.isAllPartitions)
- None
- else
+ if (offsetFetchRequest.isAllPartitions) {
+ val (error, allPartitionData) = coordinator.handleFetchOffsets(offsetFetchRequest.groupId)
+ if (error != Errors.NONE)
+ offsetFetchRequest.getErrorResponse(error)
+ else {
+ // clients are not allowed to see offsets for topics that are not authorized for Describe
+ val authorizedPartitionData = allPartitionData.filter { case (topicPartition, _) => authorizeTopicDescribe(topicPartition) }
+ new OffsetFetchResponse(Errors.NONE, authorizedPartitionData.asJava, header.apiVersion)
+ }
+ } else {
+ val (authorizedPartitions, unauthorizedPartitions) = offsetFetchRequest.partitions.asScala
+ .partition(authorizeTopicDescribe)
+ val (error, authorizedPartitionData) = coordinator.handleFetchOffsets(offsetFetchRequest.groupId,
Some(authorizedPartitions))
-
- // Note that we do not need to filter the partitions in the
- // metadata cache as the topic partitions will be filtered
- // in coordinator's offset manager through the offset cache
- if (header.apiVersion == 1) {
- val authorizedStatus =
- if (offsets._1 != Errors.NONE) {
- authorizedPartitions.map { partition =>
- (partition, new OffsetFetchResponse.PartitionData(
- OffsetFetchResponse.INVALID_OFFSET, OffsetFetchResponse.NO_METADATA, offsets._1))}.toMap
- }
- else
- offsets._2.toMap
- new OffsetFetchResponse(Errors.NONE, (authorizedStatus ++ unauthorizedStatus).asJava, header.apiVersion)
- }
- else if (offsets._1 == Errors.NONE) {
- if (offsetFetchRequest.isAllPartitions) {
- // filter out unauthorized topics in case all group offsets are requested
- val authorizedStatus = offsets._2.filter {
- case (partition, _) => authorize(request.session, Describe, new Resource(auth.Topic, partition.topic))
- }
- new OffsetFetchResponse((authorizedStatus).asJava)
+ if (error != Errors.NONE)
+ offsetFetchRequest.getErrorResponse(error)
+ else {
+ val unauthorizedPartitionData = unauthorizedPartitions.map(_ -> OffsetFetchResponse.UNKNOWN_PARTITION).toMap
+ new OffsetFetchResponse(Errors.NONE, (authorizedPartitionData ++ unauthorizedPartitionData).asJava, header.apiVersion)
}
- else
- new OffsetFetchResponse((offsets._2.toMap ++ unauthorizedStatus).asJava)
}
- else
- new OffsetFetchResponse(offsets._1)
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a84b241/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
index 42251fa..d43d1af 100644
--- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
@@ -40,6 +40,7 @@ import scala.collection.mutable
import scala.collection.mutable.Buffer
import org.apache.kafka.common.KafkaException
import kafka.admin.AdminUtils
+import kafka.network.SocketServer
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.record.MemoryRecords
@@ -173,7 +174,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
producers.foreach(_.close())
consumers.foreach(_.wakeup())
consumers.foreach(_.close())
- removeAllAcls
+ removeAllAcls()
super.tearDown()
}
@@ -712,6 +713,34 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
}
@Test
+ def testFetchAllOffsetsTopicAuthorization() {
+ val offset = 15L
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), groupResource)
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), topicResource)
+ this.consumers.head.assign(List(tp).asJava)
+ this.consumers.head.commitSync(Map(tp -> new OffsetAndMetadata(offset)).asJava)
+
+ removeAllAcls()
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), groupResource)
+
+ // send offset fetch requests directly since the consumer does not expose an API to do so
+ // note there's only one broker, so no need to lookup the group coordinator
+
+ // without describe permission on the topic, we shouldn't be able to fetch offsets
+ val offsetFetchRequest = requests.OffsetFetchRequest.forAllPartitions(group)
+ var offsetFetchResponse = sendOffsetFetchRequest(offsetFetchRequest, anySocketServer)
+ assertEquals(Errors.NONE, offsetFetchResponse.error)
+ assertTrue(offsetFetchResponse.responseData.isEmpty)
+
+ // now add describe permission on the topic and verify that the offset can be fetched
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Describe)), topicResource)
+ offsetFetchResponse = sendOffsetFetchRequest(offsetFetchRequest, anySocketServer)
+ assertEquals(Errors.NONE, offsetFetchResponse.error)
+ assertTrue(offsetFetchResponse.responseData.containsKey(tp))
+ assertEquals(offset, offsetFetchResponse.responseData.get(tp).offset)
+ }
+
+ @Test
def testOffsetFetchTopicDescribe() {
addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), groupResource)
addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Describe)), topicResource)
@@ -846,4 +875,10 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
}
}
+ private def sendOffsetFetchRequest(request: requests.OffsetFetchRequest,
+ socketServer: SocketServer): requests.OffsetFetchResponse = {
+ val response = send(request, ApiKeys.OFFSET_FETCH, socketServer)
+ requests.OffsetFetchResponse.parse(response, request.version)
+ }
+
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a84b241/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala b/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala
index 20e512f..d3de16d 100644
--- a/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala
@@ -25,7 +25,7 @@ import kafka.server.{DelayedOperationPurgatory, KafkaConfig, ReplicaManager}
import kafka.utils._
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.requests.{JoinGroupRequest, OffsetCommitRequest}
+import org.apache.kafka.common.requests.{JoinGroupRequest, OffsetCommitRequest, OffsetFetchResponse}
import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
import org.easymock.{Capture, EasyMock, IAnswer}
import org.junit.{After, Before, Test}
@@ -749,6 +749,62 @@ class GroupCoordinatorResponseTest extends JUnitSuite {
}
@Test
+ def testFetchOffsets() {
+ val tp = new TopicPartition("topic", 0)
+ val offset = OffsetAndMetadata(0)
+
+ val commitOffsetResult = commitOffsets(groupId, OffsetCommitRequest.DEFAULT_MEMBER_ID,
+ OffsetCommitRequest.DEFAULT_GENERATION_ID, immutable.Map(tp -> offset))
+ assertEquals(Errors.NONE.code, commitOffsetResult(tp))
+
+ val (error, partitionData) = groupCoordinator.handleFetchOffsets(groupId, Some(Seq(tp)))
+ assertEquals(Errors.NONE, error)
+ assertEquals(Some(0), partitionData.get(tp).map(_.offset))
+ }
+
+ @Test
+ def testFetchOffsetForUnknownPartition(): Unit = {
+ val tp = new TopicPartition("topic", 0)
+ val (error, partitionData) = groupCoordinator.handleFetchOffsets(groupId, Some(Seq(tp)))
+ assertEquals(Errors.NONE, error)
+ assertEquals(Some(OffsetFetchResponse.INVALID_OFFSET), partitionData.get(tp).map(_.offset))
+ }
+
+ @Test
+ def testFetchOffsetNotCoordinatorForGroup(): Unit = {
+ val tp = new TopicPartition("topic", 0)
+ val (error, partitionData) = groupCoordinator.handleFetchOffsets(otherGroupId, Some(Seq(tp)))
+ assertEquals(Errors.NOT_COORDINATOR_FOR_GROUP, error)
+ assertTrue(partitionData.isEmpty)
+ }
+
+ @Test
+ def testFetchAllOffsets() {
+ val tp1 = new TopicPartition("topic", 0)
+ val tp2 = new TopicPartition("topic", 1)
+ val tp3 = new TopicPartition("other-topic", 0)
+ val offset1 = OffsetAndMetadata(15)
+ val offset2 = OffsetAndMetadata(16)
+ val offset3 = OffsetAndMetadata(17)
+
+ assertEquals((Errors.NONE, Map.empty), groupCoordinator.handleFetchOffsets(groupId))
+
+ val commitOffsetResult = commitOffsets(groupId, OffsetCommitRequest.DEFAULT_MEMBER_ID,
+ OffsetCommitRequest.DEFAULT_GENERATION_ID, immutable.Map(tp1 -> offset1, tp2 -> offset2, tp3 -> offset3))
+ assertEquals(Errors.NONE.code, commitOffsetResult(tp1))
+ assertEquals(Errors.NONE.code, commitOffsetResult(tp2))
+ assertEquals(Errors.NONE.code, commitOffsetResult(tp3))
+
+ val (error, partitionData) = groupCoordinator.handleFetchOffsets(groupId)
+ assertEquals(Errors.NONE, error)
+ assertEquals(3, partitionData.size)
+ assertTrue(partitionData.forall(_._2.error == Errors.NONE))
+ assertEquals(Some(offset1.offset), partitionData.get(tp1).map(_.offset))
+ assertEquals(Some(offset2.offset), partitionData.get(tp2).map(_.offset))
+ assertEquals(Some(offset3.offset), partitionData.get(tp3).map(_.offset))
+ }
+
+ @Test
def testCommitOffsetInAwaitingSync() {
val memberId = JoinGroupRequest.UNKNOWN_MEMBER_ID
val tp = new TopicPartition("topic", 0)