From 9d0877e074d41fdf49cf1ad4d94fc53628209f1c Mon Sep 17 00:00:00 2001 From: David Jacot Date: Tue, 11 Apr 2023 10:12:07 +0200 Subject: [PATCH 1/3] Add ClientAssignor, Assignment, TopicMetadata and VersionedMetadata --- .../group/consumer/Assignment.java | 144 +++++++++++++++++ .../group/consumer/ClientAssignor.java | 150 ++++++++++++++++++ .../group/consumer/TopicMetadata.java | 109 +++++++++++++ .../group/consumer/VersionedMetadata.java | 85 ++++++++++ .../group/consumer/AssignmentTest.java | 137 ++++++++++++++++ .../group/consumer/AssignmentTestUtil.java | 68 ++++++++ .../group/consumer/ClientAssignorTest.java | 123 ++++++++++++++ .../group/consumer/TopicMetadataTest.java | 66 ++++++++ .../group/consumer/VersionedMetadataTest.java | 60 +++++++ 9 files changed, 942 insertions(+) create mode 100644 group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/Assignment.java create mode 100644 group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ClientAssignor.java create mode 100644 group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java create mode 100644 group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadata.java create mode 100644 group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTest.java create mode 100644 group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTestUtil.java create mode 100644 group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java create mode 100644 group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TopicMetadataTest.java create mode 100644 group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadataTest.java diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/Assignment.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/Assignment.java new file mode 100644 index 0000000000000..9cf6b521dd26f --- /dev/null +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/Assignment.java @@ -0,0 +1,144 @@ +/* + * 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.coordinator.group.consumer; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.coordinator.group.generated.ConsumerGroupTargetAssignmentMemberValue; + +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.HashSet; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * An immutable assignment for a member. + */ +public class Assignment { + public static final Assignment EMPTY = new Assignment( + (byte) 0, + Collections.emptyMap(), + VersionedMetadata.EMPTY + ); + + /** + * The error assigned to the member. + */ + private final byte error; + + /** + * The partitions assigned to the member. + */ + private final Map> partitions; + + /** + * The metadata assigned to the member. + */ + private final VersionedMetadata metadata; + + public Assignment( + Map> partitions + ) { + this( + (byte) 0, + partitions, + VersionedMetadata.EMPTY + ); + } + + public Assignment( + byte error, + Map> partitions, + VersionedMetadata metadata + ) { + this.error = error; + this.partitions = Collections.unmodifiableMap(Objects.requireNonNull(partitions)); + this.metadata = Objects.requireNonNull(metadata); + } + + /** + * @return The error. + */ + public byte error() { + return error; + } + + /** + * @return The assigned partitions. + */ + public Map> partitions() { + return partitions; + } + + /** + * @return The metadata. + */ + public VersionedMetadata metadata() { + return metadata; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + Assignment that = (Assignment) o; + + if (error != that.error) return false; + if (!partitions.equals(that.partitions)) return false; + return metadata.equals(that.metadata); + } + + @Override + public int hashCode() { + int result = error; + result = 31 * result + partitions.hashCode(); + result = 31 * result + metadata.hashCode(); + return result; + } + + @Override + public String toString() { + return "Assignment(" + + "error=" + error + + ", partitions=" + partitions + + ", metadata=" + metadata + + ')'; + } + + /** + * Creates a {{@link Assignment}} from a {{@link ConsumerGroupTargetAssignmentMemberValue}}. + * + * @param record The record. + * @return A {{@link Assignment}}. + */ + public static Assignment fromRecord( + ConsumerGroupTargetAssignmentMemberValue record + ) { + return new Assignment( + record.error(), + record.topicPartitions().stream().collect(Collectors.toMap( + ConsumerGroupTargetAssignmentMemberValue.TopicPartition::topicId, + topicPartitions -> new HashSet<>(topicPartitions.partitions()))), + new VersionedMetadata( + record.metadataVersion(), + ByteBuffer.wrap(record.metadataBytes())) + ); + } +} diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ClientAssignor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ClientAssignor.java new file mode 100644 index 0000000000000..26fded3ac01e1 --- /dev/null +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ClientAssignor.java @@ -0,0 +1,150 @@ +/* + * 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.coordinator.group.consumer; + +import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataValue; + +import java.nio.ByteBuffer; +import java.util.Objects; + +/** + * An immutable representation of a client side assignor within a consumer group member. + */ +public class ClientAssignor { + /** + * The name of the assignor. + */ + private final String name; + + /** + * The reason reported by the assignor. + */ + private final byte reason; + + /** + * The minimum metadata version supported by the assignor. + */ + private final short minimumVersion; + + /** + * The maximum metadata version supported by the assignor. + */ + private final short maximumVersion; + + /** + * The versioned metadata. + */ + private final VersionedMetadata metadata; + + public ClientAssignor( + String name, + byte reason, + short minimumVersion, + short maximumVersion, + VersionedMetadata metadata + ) { + this.name = Objects.requireNonNull(name); + this.reason = reason; + this.minimumVersion = minimumVersion; + this.maximumVersion = maximumVersion; + this.metadata = Objects.requireNonNull(metadata); + } + + /** + * @return The client side assignor name. + */ + public String name() { + return this.name; + } + + /** + * @return The current reason reported by the assignor. + */ + public byte reason() { + return this.reason; + } + + /** + * @return The minimum version supported by the assignor. + */ + public short minimumVersion() { + return this.minimumVersion; + } + + /** + * @return The maximum version supported by the assignor. + */ + public short maximumVersion() { + return this.maximumVersion; + } + + /** + * @return The versioned metadata. + */ + public VersionedMetadata metadata() { + return metadata; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + ClientAssignor that = (ClientAssignor) o; + + if (reason != that.reason) return false; + if (minimumVersion != that.minimumVersion) return false; + if (maximumVersion != that.maximumVersion) return false; + if (!name.equals(that.name)) return false; + return metadata.equals(that.metadata); + } + + @Override + public int hashCode() { + int result = name.hashCode(); + result = 31 * result + (int) reason; + result = 31 * result + (int) minimumVersion; + result = 31 * result + (int) maximumVersion; + result = 31 * result + metadata.hashCode(); + return result; + } + + @Override + public String toString() { + return "ClientAssignor(name=" + name + + ", reason=" + reason + + ", minimumVersion=" + minimumVersion + + ", maximumVersion=" + maximumVersion + + ", metadata=" + metadata + + ')'; + } + + public static ClientAssignor fromRecord( + ConsumerGroupMemberMetadataValue.Assignor record + ) { + return new ClientAssignor( + record.name(), + record.reason(), + record.minimumVersion(), + record.maximumVersion(), + new VersionedMetadata( + record.version(), + ByteBuffer.wrap(record.metadata()) + ) + ); + } +} diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java new file mode 100644 index 0000000000000..1b638eee73919 --- /dev/null +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java @@ -0,0 +1,109 @@ +/* + * 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.coordinator.group.consumer; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.coordinator.group.generated.ConsumerGroupPartitionMetadataValue; + +import java.util.Objects; + +public class TopicMetadata { + /** + * The topic id. + */ + private final Uuid id; + + /** + * The topic name. + */ + private final String name; + + /** + * The number of partitions. + */ + private final int numPartitions; + + public TopicMetadata( + Uuid id, + String name, + int numPartitions + ) { + this.id = Objects.requireNonNull(id); + this.name = Objects.requireNonNull(name); + this.numPartitions = numPartitions; + } + + /** + * @return The topic id. + */ + public Uuid id() { + return this.id; + } + + /** + * @return The topic name. + */ + public String name() { + return this.name; + } + + /** + * @return The number of partitions. + */ + public int numPartitions() { + return this.numPartitions; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + TopicMetadata that = (TopicMetadata) o; + + if (!id.equals(that.id)) return false; + if (!name.equals(that.name)) return false; + return numPartitions == that.numPartitions; + } + + @Override + public int hashCode() { + int result = id.hashCode(); + result = 31 * result + name.hashCode(); + result = 31 * result + numPartitions; + return result; + } + + @Override + public String toString() { + return "TopicMetadata(" + + "id=" + id + + ", name=" + name + + ", numPartitions=" + numPartitions + + ')'; + } + + public static TopicMetadata fromRecord( + ConsumerGroupPartitionMetadataValue.TopicMetadata record + ) { + return new TopicMetadata( + record.topicId(), + record.topicName(), + record.numPartitions() + ); + } +} diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadata.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadata.java new file mode 100644 index 0000000000000..32ac4c362beed --- /dev/null +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadata.java @@ -0,0 +1,85 @@ +/* + * 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.coordinator.group.consumer; + +import java.nio.ByteBuffer; +import java.util.Objects; + +/** + * Immutable versioned metadata. + */ +public class VersionedMetadata { + public static final VersionedMetadata EMPTY = new VersionedMetadata((short) 0, ByteBuffer.allocate(0)); + + /** + * The version of the metadata encoded in {{@link VersionedMetadata#metadata}}. + */ + private final short version; + + /** + * The metadata bytes. + */ + private final ByteBuffer metadata; + + public VersionedMetadata( + short version, + ByteBuffer metadata + ) { + this.version = version; + this.metadata = Objects.requireNonNull(metadata); + } + + /** + * @return The version of the metadata. + */ + public short version() { + return this.version; + } + + /** + * @return The ByteBuffer holding the metadata. + */ + public ByteBuffer metadata() { + return this.metadata; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + VersionedMetadata that = (VersionedMetadata) o; + + if (version != that.version) return false; + return metadata.equals(that.metadata); + } + + @Override + public int hashCode() { + int result = version; + result = 31 * result + metadata.hashCode(); + return result; + } + + @Override + public String toString() { + return "VersionedMetadata(" + + "version=" + version + + ", metadata=" + metadata + + ')'; + } +} diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTest.java new file mode 100644 index 0000000000000..ca3292d85ef9f --- /dev/null +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTest.java @@ -0,0 +1,137 @@ +/* + * 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.coordinator.group.consumer; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.coordinator.group.generated.ConsumerGroupTargetAssignmentMemberValue; +import org.junit.jupiter.api.Test; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.apache.kafka.coordinator.group.consumer.AssignmentTestUtil.mkAssignment; +import static org.apache.kafka.coordinator.group.consumer.AssignmentTestUtil.mkTopicAssignment; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class AssignmentTest { + + @Test + public void testPartitionsAndMetadataCannotBeNull() { + assertThrows(NullPointerException.class, () -> new Assignment( + (byte) 1, + null, + new VersionedMetadata( + (short) 1, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ) + )); + + assertThrows(NullPointerException.class, () -> new Assignment( + (byte) 1, + mkAssignment( + mkTopicAssignment(Uuid.randomUuid(), 1, 2, 3) + ), + null + )); + } + + @Test + public void testAttributes() { + Map> partitions = mkAssignment( + mkTopicAssignment(Uuid.randomUuid(), 1, 2, 3) + ); + + VersionedMetadata metadata = new VersionedMetadata( + (short) 1, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ); + + Assignment assignment = new Assignment( + (byte) 1, + partitions, + metadata + ); + + assertEquals((byte) 1, assignment.error()); + assertEquals(partitions, assignment.partitions()); + assertEquals(metadata, assignment.metadata()); + } + + @Test + public void testFromTargetAssignmentRecord() { + Uuid topicId1 = Uuid.randomUuid(); + Uuid topicId2 = Uuid.randomUuid(); + + List partitions = new ArrayList<>(); + partitions.add(new ConsumerGroupTargetAssignmentMemberValue.TopicPartition() + .setTopicId(topicId1) + .setPartitions(Arrays.asList(1, 2, 3))); + partitions.add(new ConsumerGroupTargetAssignmentMemberValue.TopicPartition() + .setTopicId(topicId2) + .setPartitions(Arrays.asList(4, 5, 6))); + + ConsumerGroupTargetAssignmentMemberValue record = new ConsumerGroupTargetAssignmentMemberValue() + .setError((byte) 1) + .setTopicPartitions(partitions) + .setMetadataVersion((short) 2) + .setMetadataBytes("foo".getBytes(StandardCharsets.UTF_8)); + + Assignment assignment = Assignment.fromRecord(record); + + assertEquals((short) 1, assignment.error()); + assertEquals(mkAssignment( + mkTopicAssignment(topicId1, 1, 2, 3), + mkTopicAssignment(topicId2, 4, 5, 6) + ), assignment.partitions()); + assertEquals(new VersionedMetadata( + (short) 2, + ByteBuffer.wrap("foo".getBytes(StandardCharsets.UTF_8)) + ), assignment.metadata()); + } + + @Test + public void testEquals() { + Map> partitions = mkAssignment( + mkTopicAssignment(Uuid.randomUuid(), 1, 2, 3) + ); + + VersionedMetadata metadata = new VersionedMetadata( + (short) 1, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ); + + Assignment assignment1 = new Assignment( + (byte) 1, + partitions, + metadata + ); + + Assignment assignment2 = new Assignment( + (byte) 1, + partitions, + metadata + ); + + assertEquals(assignment1, assignment2); + } +} diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTestUtil.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTestUtil.java new file mode 100644 index 0000000000000..8b5e072aaa05e --- /dev/null +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTestUtil.java @@ -0,0 +1,68 @@ +/* + * 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.coordinator.group.consumer; + +import org.apache.kafka.common.Uuid; + +import java.util.AbstractMap; +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; + +public class AssignmentTestUtil { + public static Map.Entry> mkTopicAssignment( + Uuid topicId, + Integer... partitions + ) { + return new AbstractMap.SimpleEntry<>( + topicId, + new HashSet<>(Arrays.asList(partitions)) + ); + } + + public static Map.Entry> mkSortedTopicAssignment( + Uuid topicId, + Integer... partitions + ) { + return new AbstractMap.SimpleEntry<>( + topicId, + new TreeSet<>(Arrays.asList(partitions)) + ); + } + + @SafeVarargs + public static Map> mkAssignment(Map.Entry>... entries) { + Map> assignment = new HashMap<>(); + for (Map.Entry> entry : entries) { + assignment.put(entry.getKey(), entry.getValue()); + } + return assignment; + } + + @SafeVarargs + public static Map> mkSortedAssignment(Map.Entry>... entries) { + Map> assignment = new LinkedHashMap<>(); + for (Map.Entry> entry : entries) { + assignment.put(entry.getKey(), entry.getValue()); + } + return assignment; + } +} diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java new file mode 100644 index 0000000000000..d91b56e8e361d --- /dev/null +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java @@ -0,0 +1,123 @@ +/* + * 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.coordinator.group.consumer; + +import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataValue; +import org.junit.jupiter.api.Test; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class ClientAssignorTest { + + @Test + public void testNameAndMetadataCannotBeBull() { + assertThrows(NullPointerException.class, () -> new ClientAssignor( + "range", + (byte) 2, + (short) 5, + (short) 10, + null + )); + + assertThrows(NullPointerException.class, () -> new ClientAssignor( + null, + (byte) 2, + (short) 5, + (short) 10, + new VersionedMetadata( + (short) 8, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ) + )); + } + + @Test + public void testAttributes() { + ClientAssignor clientAssignor = new ClientAssignor( + "range", + (byte) 2, + (short) 5, + (short) 10, + new VersionedMetadata( + (short) 8, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ) + ); + + assertEquals("range", clientAssignor.name()); + assertEquals((byte) 2, clientAssignor.reason()); + assertEquals((short) 5, clientAssignor.minimumVersion()); + assertEquals((short) 10, clientAssignor.maximumVersion()); + assertEquals(new VersionedMetadata( + (short) 8, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ), clientAssignor.metadata()); + } + + @Test + public void testFromRecord() { + ConsumerGroupMemberMetadataValue.Assignor record = new ConsumerGroupMemberMetadataValue.Assignor() + .setName("range") + .setReason((byte) 2) + .setMinimumVersion((byte) 5) + .setMaximumVersion((byte) 10) + .setVersion((byte) 8) + .setMetadata("hello".getBytes(StandardCharsets.UTF_8)); + + ClientAssignor clientAssignor = ClientAssignor.fromRecord(record); + + assertEquals("range", clientAssignor.name()); + assertEquals((byte) 2, clientAssignor.reason()); + assertEquals((short) 5, clientAssignor.minimumVersion()); + assertEquals((short) 10, clientAssignor.maximumVersion()); + assertEquals(new VersionedMetadata( + (short) 8, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ), clientAssignor.metadata()); + } + + @Test + public void testEquals() { + ClientAssignor clientAssignor1 = new ClientAssignor( + "range", + (byte) 2, + (short) 5, + (short) 10, + new VersionedMetadata( + (short) 8, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ) + ); + + ClientAssignor clientAssignor2 = new ClientAssignor( + "range", + (byte) 2, + (short) 5, + (short) 10, + new VersionedMetadata( + (short) 8, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ) + ); + + assertEquals(clientAssignor1, clientAssignor2); + } +} diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TopicMetadataTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TopicMetadataTest.java new file mode 100644 index 0000000000000..07d790ec8ff93 --- /dev/null +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TopicMetadataTest.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.coordinator.group.consumer; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.coordinator.group.generated.ConsumerGroupPartitionMetadataValue; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class TopicMetadataTest { + @Test + public void testAttributes() { + Uuid topicId = Uuid.randomUuid(); + TopicMetadata topicMetadata = new TopicMetadata(topicId, "foo", 15); + assertEquals(topicId, topicMetadata.id()); + assertEquals("foo", topicMetadata.name()); + assertEquals(15, topicMetadata.numPartitions()); + } + + @Test + public void testTopicIdAndNameCannotBeNull() { + assertThrows(NullPointerException.class, () -> new TopicMetadata(Uuid.randomUuid(), null, 15)); + assertThrows(NullPointerException.class, () -> new TopicMetadata(null, "foo", 15)); + } + + @Test + public void testEquals() { + Uuid topicId = Uuid.randomUuid(); + TopicMetadata topicMetadata = new TopicMetadata(topicId, "foo", 15); + assertEquals(new TopicMetadata(topicId, "foo", 15), topicMetadata); + assertNotEquals(new TopicMetadata(topicId, "foo", 5), topicMetadata); + } + + @Test + public void testFromRecord() { + Uuid topicId = Uuid.randomUuid(); + String topicName = "foo"; + + ConsumerGroupPartitionMetadataValue.TopicMetadata record = new ConsumerGroupPartitionMetadataValue.TopicMetadata() + .setTopicId(topicId) + .setTopicName(topicName) + .setNumPartitions(15); + + assertEquals( + new TopicMetadata(topicId, topicName, 15), + TopicMetadata.fromRecord(record) + ); + } +} diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadataTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadataTest.java new file mode 100644 index 0000000000000..ab6542a8e6565 --- /dev/null +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadataTest.java @@ -0,0 +1,60 @@ +/* + * 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.coordinator.group.consumer; + +import org.junit.jupiter.api.Test; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class VersionedMetadataTest { + @Test + public void testAttributes() { + VersionedMetadata metadata = new VersionedMetadata( + (short) 1, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ); + + assertEquals((short) 1, metadata.version()); + assertEquals(ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)), metadata.metadata()); + } + + @Test + public void testMetadataCannotBeNull() { + assertThrows(NullPointerException.class, () -> new VersionedMetadata((short) 1, null)); + } + + @Test + public void testEquals() { + VersionedMetadata metadata1 = new VersionedMetadata( + (short) 1, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ); + + VersionedMetadata metadata2 = new VersionedMetadata( + (short) 1, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ); + + assertEquals(metadata1, metadata2); + assertNotEquals(metadata1, VersionedMetadata.EMPTY); + } +} From bad3e03c12e21b6a9241fd277a74c5adfdf5d75a Mon Sep 17 00:00:00 2001 From: David Jacot Date: Thu, 20 Apr 2023 10:27:29 +0200 Subject: [PATCH 2/3] address comments --- .../group/consumer/ClientAssignor.java | 14 ++++++++++++++ .../group/consumer/TopicMetadata.java | 12 ++++++++++++ .../group/consumer/VersionedMetadata.java | 4 +++- .../group/consumer/AssignmentTest.java | 14 ++++++++++---- .../group/consumer/ClientAssignorTest.java | 18 ++++++++++++++---- .../group/consumer/VersionedMetadataTest.java | 9 ++++----- 6 files changed, 57 insertions(+), 14 deletions(-) diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ClientAssignor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ClientAssignor.java index 26fded3ac01e1..fed73cd58d39f 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ClientAssignor.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ClientAssignor.java @@ -58,9 +58,23 @@ public ClientAssignor( VersionedMetadata metadata ) { this.name = Objects.requireNonNull(name); + if (name.isEmpty()) { + throw new IllegalArgumentException("Assignor name cannot be empty."); + } this.reason = reason; this.minimumVersion = minimumVersion; + if (minimumVersion < -1) { + // -1 is supported as part of the upgrade from the old protocol to the new protocol. It + // basically means that the assignor supports metadata from the old client assignor. + throw new IllegalArgumentException("Assignor minimum version must be greater than -1."); + } this.maximumVersion = maximumVersion; + if (maximumVersion < 0) { + throw new IllegalArgumentException("Assignor maximum version must be greater than or equals to 0."); + } else if (maximumVersion < minimumVersion) { + throw new IllegalArgumentException("Assignor maximum version must be greater than or equals to " + + "the minimum version."); + } this.metadata = Objects.requireNonNull(metadata); } diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java index 1b638eee73919..395d48d95dffb 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TopicMetadata.java @@ -21,6 +21,9 @@ import java.util.Objects; +/** + * Immutable topic metadata. + */ public class TopicMetadata { /** * The topic id. @@ -43,8 +46,17 @@ public TopicMetadata( int numPartitions ) { this.id = Objects.requireNonNull(id); + if (Uuid.ZERO_UUID.equals(id)) { + throw new IllegalArgumentException("Topic id cannot be ZERO_UUID."); + } this.name = Objects.requireNonNull(name); + if (name.isEmpty()) { + throw new IllegalArgumentException("Topic name cannot be empty."); + } this.numPartitions = numPartitions; + if (numPartitions < 0) { + throw new IllegalArgumentException("Number of partitions cannot be negative."); + } } /** diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadata.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadata.java index 32ac4c362beed..89baeb82bfa6d 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadata.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadata.java @@ -20,7 +20,9 @@ import java.util.Objects; /** - * Immutable versioned metadata. + * Immutable versioned metadata. It contains a bunch of bytes tagged with a version. The + * format of the bytes is unspecified. This is mainly used by client side assignors to + * exchange arbitrary metadata between the members and the assignor and vice versa. */ public class VersionedMetadata { public static final VersionedMetadata EMPTY = new VersionedMetadata((short) 0, ByteBuffer.allocate(0)); diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTest.java index ca3292d85ef9f..d81006cc026e0 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/AssignmentTest.java @@ -24,6 +24,7 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Set; @@ -31,6 +32,7 @@ import static org.apache.kafka.coordinator.group.consumer.AssignmentTestUtil.mkAssignment; import static org.apache.kafka.coordinator.group.consumer.AssignmentTestUtil.mkTopicAssignment; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertThrows; public class AssignmentTest { @@ -120,18 +122,22 @@ public void testEquals() { ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) ); - Assignment assignment1 = new Assignment( + Assignment assignment = new Assignment( (byte) 1, partitions, metadata ); - Assignment assignment2 = new Assignment( + assertEquals(new Assignment( (byte) 1, partitions, metadata - ); + ), assignment); - assertEquals(assignment1, assignment2); + assertNotEquals(new Assignment( + (byte) 1, + Collections.emptyMap(), + metadata + ), assignment); } } diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java index d91b56e8e361d..5abe8b129f86f 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java @@ -23,6 +23,7 @@ import java.nio.charset.StandardCharsets; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertThrows; public class ClientAssignorTest { @@ -96,7 +97,7 @@ public void testFromRecord() { @Test public void testEquals() { - ClientAssignor clientAssignor1 = new ClientAssignor( + ClientAssignor clientAssignor = new ClientAssignor( "range", (byte) 2, (short) 5, @@ -107,7 +108,7 @@ public void testEquals() { ) ); - ClientAssignor clientAssignor2 = new ClientAssignor( + assertEquals(new ClientAssignor( "range", (byte) 2, (short) 5, @@ -116,8 +117,17 @@ public void testEquals() { (short) 8, ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) ) - ); + ), clientAssignor); - assertEquals(clientAssignor1, clientAssignor2); + assertNotEquals(new ClientAssignor( + "uniform", + (byte) 2, + (short) 5, + (short) 10, + new VersionedMetadata( + (short) 8, + ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) + ) + ), clientAssignor); } } diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadataTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadataTest.java index ab6542a8e6565..10997d015cb63 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadataTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/VersionedMetadataTest.java @@ -44,17 +44,16 @@ public void testMetadataCannotBeNull() { @Test public void testEquals() { - VersionedMetadata metadata1 = new VersionedMetadata( + VersionedMetadata metadata = new VersionedMetadata( (short) 1, ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) ); - VersionedMetadata metadata2 = new VersionedMetadata( + assertEquals(new VersionedMetadata( (short) 1, ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)) - ); + ), metadata); - assertEquals(metadata1, metadata2); - assertNotEquals(metadata1, VersionedMetadata.EMPTY); + assertNotEquals(VersionedMetadata.EMPTY, metadata); } } From f642b451f23903e29cbae7be281fccd7c1fc81f1 Mon Sep 17 00:00:00 2001 From: David Jacot Date: Thu, 20 Apr 2023 20:22:32 +0200 Subject: [PATCH 3/3] fix typo --- .../kafka/coordinator/group/consumer/ClientAssignorTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java index 5abe8b129f86f..d136a411cd8ec 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ClientAssignorTest.java @@ -29,7 +29,7 @@ public class ClientAssignorTest { @Test - public void testNameAndMetadataCannotBeBull() { + public void testNameAndMetadataCannotBeNull() { assertThrows(NullPointerException.class, () -> new ClientAssignor( "range", (byte) 2,