You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2024/08/19 16:10:14 UTC
[PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
smjn opened a new pull request, #16921:
URL: https://github.com/apache/kafka/pull/16921
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #16921:
URL: https://github.com/apache/kafka/pull/16921#issuecomment-2298255030
> Thanks for the PR. I would expect that there will be some checkstyle problems in the build because of the indentation.
Not seeing any issues - neither the infra nor locally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1726384651
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
+
+ /**
+ * Return the partition index for the given key.
+ *
+ * @param key key - groupId:topicId:partitionId.
+ * @return The partition index.
+ */
+ int partitionFor(String key);
+
+ /**
+ * Return the configuration properties of the share-group state topic.
+ *
+ * @return Properties of the share-group state topic.
+ */
+ Properties shareGroupStateTopicConfigs();
+
+ /**
+ * Start the share coordinator
+ *
+ * @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
+ */
+ void startup(IntSupplier shareGroupTopicPartitionCount);
+
+ /**
+ * Stop the share coordinator
+ */
+ void shutdown();
+
+ /**
+ * Handle write share state call
+ * @param context - represents the incoming write request context
Review Comment:
Following
```
* @param key key - groupId:topicId:partitionId.
* @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
* @param context - represents the incoming write request context
* @param request - actual RPC request object
* @param context - represents the incoming read request context
* @param request - actual RPC request object
* @param partitionIndex - The partition index (internal topic)
* @param partitionLeaderEpoch - Leader epoch of the partition (internal topic)
* @param partitionIndex - The partition index (internal topic)
* @param partitionLeaderEpoch - Leader epoch of the partition (internal topic). Empty optional means deleted.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727207040
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterOffsetsStateBatch.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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.Objects;
+
+/**
+ * This is a helper class which overrides the equals and hashcode
+ * methods to only focus on the first and last offset fields of the
+ * state batch. This is useful when combining batches.
+ */
+public class PersisterOffsetsStateBatch extends PersisterStateBatch {
+ public PersisterOffsetsStateBatch(long firstOffset, long lastOffset, byte deliveryState, short deliveryCount) {
+ super(firstOffset, lastOffset, deliveryState, deliveryCount);
+ }
+
+ public static PersisterOffsetsStateBatch of(ShareSnapshotValue.StateBatch batch) {
+ return new PersisterOffsetsStateBatch(batch.firstOffset(), batch.lastOffset(), batch.deliveryState(), batch.deliveryCount());
+ }
+
+ public static PersisterOffsetsStateBatch of(ShareUpdateValue.StateBatch batch) {
+ return new PersisterOffsetsStateBatch(batch.firstOffset(), batch.lastOffset(), batch.deliveryState(), batch.deliveryCount());
+ }
+
+ public static PersisterOffsetsStateBatch of(PersisterStateBatch batch) {
+ return new PersisterOffsetsStateBatch(batch.firstOffset(), batch.lastOffset(), batch.deliveryState(), batch.deliveryCount());
+ }
+
+ public static PersisterOffsetsStateBatch of(WriteShareGroupStateRequestData.StateBatch batch) {
+ return new PersisterOffsetsStateBatch(batch.firstOffset(), batch.lastOffset(), batch.deliveryState(), batch.deliveryCount());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof PersisterOffsetsStateBatch)) {
+ return false;
+ }
+ PersisterOffsetsStateBatch that = (PersisterOffsetsStateBatch) o;
+ return this.firstOffset() == that.firstOffset() && this.lastOffset() == that.lastOffset();
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(firstOffset(), lastOffset());
+ }
+}
Review Comment:
nit: missing newline here and in other files
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1726384651
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
+
+ /**
+ * Return the partition index for the given key.
+ *
+ * @param key key - groupId:topicId:partitionId.
+ * @return The partition index.
+ */
+ int partitionFor(String key);
+
+ /**
+ * Return the configuration properties of the share-group state topic.
+ *
+ * @return Properties of the share-group state topic.
+ */
+ Properties shareGroupStateTopicConfigs();
+
+ /**
+ * Start the share coordinator
+ *
+ * @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
+ */
+ void startup(IntSupplier shareGroupTopicPartitionCount);
+
+ /**
+ * Stop the share coordinator
+ */
+ void shutdown();
+
+ /**
+ * Handle write share state call
+ * @param context - represents the incoming write request context
Review Comment:
Following. All occurrences in PR
```
* @param key - groupId:topicId:partitionId.
* @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
* @param context - represents the incoming write request context
* @param request - actual RPC request object
* @param context - represents the incoming read request context
* @param request - actual RPC request object
* @param partitionIndex - The partition index (internal topic)
* @param partitionLeaderEpoch - Leader epoch of the partition (internal topic)
* @param partitionIndex - The partition index (internal topic)
* @param partitionLeaderEpoch - Leader epoch of the partition (internal topic). Empty optional means deleted.
* @param recordType - short representing version
* @param recordType - short representing version
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1725687920
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ public final String groupId;
+ public final Uuid topicId;
+ public final int partition;
Review Comment:
Yeah, but if we are not using them externally then should we mark them private?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727373772
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
+
+ public SharePartitionKey(String groupId, Uuid topicId, int partition) {
+ this.groupId = groupId;
+ this.topicId = topicId;
+ this.partition = partition;
+ }
+
+ public static class Builder {
+ private String groupId;
+ private Uuid topicId;
+ private int partition;
+
+ public Builder setGroupId(String groupId) {
+ this.groupId = groupId;
+ return this;
+ }
+
+ public Builder setTopicId(Uuid topicId) {
+ this.topicId = topicId;
+ return this;
+ }
+
+ public Builder setPartition(int partition) {
+ this.partition = partition;
+ return this;
+ }
+
+ public SharePartitionKey build() {
+ return new SharePartitionKey(groupId, topicId, partition);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (!(o instanceof SharePartitionKey)) return false;
+ SharePartitionKey that = (SharePartitionKey) o;
+ return partition == that.partition && Objects.equals(groupId, that.groupId) && Objects.equals(topicId, that.topicId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(groupId, topicId, partition);
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s:%s:%d", groupId, topicId, partition);
Review Comment:
Ok, in that case lets create a separate method for the specific format we need.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727482508
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Container class to represent data encapsulated in {@link ShareSnapshotValue} and {@link ShareUpdateValue}
+ * This class is effectively immutable (state batches is not modified out of context).
+ */
+public class ShareGroupOffset {
+ private final int snapshotEpoch;
+ private final int stateEpoch;
+ private final int leaderEpoch;
+ private final long startOffset;
+ private final List<? extends PersisterStateBatch> stateBatches;
+
+ private ShareGroupOffset(int snapshotEpoch,
+ int stateEpoch,
+ int leaderEpoch,
+ long startOffset,
+ List<? extends PersisterStateBatch> stateBatches) {
+ this.snapshotEpoch = snapshotEpoch;
+ this.stateEpoch = stateEpoch;
+ this.leaderEpoch = leaderEpoch;
+ this.startOffset = startOffset;
+ this.stateBatches = stateBatches;
+ }
+
+ public int snapshotEpoch() {
+ return snapshotEpoch;
+ }
+
+ public int stateEpoch() {
+ return stateEpoch;
+ }
+
+ public int leaderEpoch() {
+ return leaderEpoch;
+ }
+
+ public long startOffset() {
+ return startOffset;
+ }
+
+ public List<? extends PersisterStateBatch> stateBatches() {
+ return Collections.unmodifiableList(stateBatches);
+ }
+
+ private static PersisterStateBatch toPersisterStateBatch(ShareSnapshotValue.StateBatch stateBatch) {
+ return new PersisterStateBatch(stateBatch.firstOffset(), stateBatch.lastOffset(), stateBatch.deliveryState(), stateBatch.deliveryCount());
+ }
+
+ private static PersisterStateBatch toPersisterStateBatch(ShareUpdateValue.StateBatch stateBatch) {
+ return new PersisterStateBatch(stateBatch.firstOffset(), stateBatch.lastOffset(), stateBatch.deliveryState(), stateBatch.deliveryCount());
+ }
+
+ public static ShareGroupOffset fromRecord(ShareSnapshotValue record) {
+ return new ShareGroupOffset(record.snapshotEpoch(), record.stateEpoch(), record.leaderEpoch(), record.startOffset(), record.stateBatches().stream()
+ .map(ShareGroupOffset::toPersisterStateBatch).collect(Collectors.toList()));
+ }
+
+ public static ShareGroupOffset fromRecord(ShareUpdateValue record) {
+ return new ShareGroupOffset(record.snapshotEpoch(), -1, record.leaderEpoch(), record.startOffset(), record.stateBatches().stream()
+ .map(ShareGroupOffset::toPersisterStateBatch).collect(Collectors.toList()));
+ }
+
+ public static ShareGroupOffset fromRequest(WriteShareGroupStateRequestData.PartitionData data) {
+ return fromRequest(data, 0);
+ }
+
+ public static ShareGroupOffset fromRequest(WriteShareGroupStateRequestData.PartitionData data, int snapshotEpoch) {
+ return new ShareGroupOffset(snapshotEpoch,
+ data.stateEpoch(),
+ data.leaderEpoch(),
+ data.startOffset(),
+ data.stateBatches().stream()
+ .map(PersisterStateBatch::from)
+ .collect(Collectors.toList()));
+ }
+
+ public Set<PersisterOffsetsStateBatch> stateBatchAsSet() {
+ return this.stateBatches().stream()
+ .map(PersisterOffsetsStateBatch::of)
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ }
+
+ public static class Builder {
+ private int snapshotEpoch;
+ private int stateEpoch;
+ private int leaderEpoch;
+ private long startOffset;
+ private List<? extends PersisterStateBatch> stateBatches;
+
+ public Builder setSnapshotEpoch(int snapshotEpoch) {
+ this.snapshotEpoch = snapshotEpoch;
+ return this;
+ }
+
+ public Builder setStateEpoch(int stateEpoch) {
+ this.stateEpoch = stateEpoch;
+ return this;
+ }
+
+ public Builder setLeaderEpoch(int leaderEpoch) {
+ this.leaderEpoch = leaderEpoch;
+ return this;
+ }
+
+ public Builder setStartOffset(long startOffset) {
+ this.startOffset = startOffset;
+ return this;
+ }
+
+ public Builder setStateBatches(List<? extends PersisterStateBatch> stateBatches) {
+ this.stateBatches = stateBatches;
+ return this;
+ }
+
+ public ShareGroupOffset build() {
+ return new ShareGroupOffset(snapshotEpoch, stateEpoch, leaderEpoch, startOffset, stateBatches);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ ShareGroupOffset that = (ShareGroupOffset) o;
+ return snapshotEpoch == that.snapshotEpoch && stateEpoch == that.stateEpoch && leaderEpoch == that.leaderEpoch && startOffset == that.startOffset && Objects.equals(stateBatches, that.stateBatches);
Review Comment:
nit: break this long line
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Container class to represent data encapsulated in {@link ShareSnapshotValue} and {@link ShareUpdateValue}
+ * This class is effectively immutable (state batches is not modified out of context).
+ */
+public class ShareGroupOffset {
+ private final int snapshotEpoch;
+ private final int stateEpoch;
+ private final int leaderEpoch;
+ private final long startOffset;
+ private final List<? extends PersisterStateBatch> stateBatches;
+
+ private ShareGroupOffset(int snapshotEpoch,
+ int stateEpoch,
+ int leaderEpoch,
+ long startOffset,
+ List<? extends PersisterStateBatch> stateBatches) {
+ this.snapshotEpoch = snapshotEpoch;
+ this.stateEpoch = stateEpoch;
+ this.leaderEpoch = leaderEpoch;
+ this.startOffset = startOffset;
+ this.stateBatches = stateBatches;
+ }
+
+ public int snapshotEpoch() {
+ return snapshotEpoch;
+ }
+
+ public int stateEpoch() {
+ return stateEpoch;
+ }
+
+ public int leaderEpoch() {
+ return leaderEpoch;
+ }
+
+ public long startOffset() {
+ return startOffset;
+ }
+
+ public List<? extends PersisterStateBatch> stateBatches() {
+ return Collections.unmodifiableList(stateBatches);
+ }
+
+ private static PersisterStateBatch toPersisterStateBatch(ShareSnapshotValue.StateBatch stateBatch) {
+ return new PersisterStateBatch(stateBatch.firstOffset(), stateBatch.lastOffset(), stateBatch.deliveryState(), stateBatch.deliveryCount());
+ }
+
+ private static PersisterStateBatch toPersisterStateBatch(ShareUpdateValue.StateBatch stateBatch) {
+ return new PersisterStateBatch(stateBatch.firstOffset(), stateBatch.lastOffset(), stateBatch.deliveryState(), stateBatch.deliveryCount());
+ }
+
+ public static ShareGroupOffset fromRecord(ShareSnapshotValue record) {
+ return new ShareGroupOffset(record.snapshotEpoch(), record.stateEpoch(), record.leaderEpoch(), record.startOffset(), record.stateBatches().stream()
+ .map(ShareGroupOffset::toPersisterStateBatch).collect(Collectors.toList()));
+ }
+
+ public static ShareGroupOffset fromRecord(ShareUpdateValue record) {
+ return new ShareGroupOffset(record.snapshotEpoch(), -1, record.leaderEpoch(), record.startOffset(), record.stateBatches().stream()
+ .map(ShareGroupOffset::toPersisterStateBatch).collect(Collectors.toList()));
+ }
+
+ public static ShareGroupOffset fromRequest(WriteShareGroupStateRequestData.PartitionData data) {
+ return fromRequest(data, 0);
+ }
+
+ public static ShareGroupOffset fromRequest(WriteShareGroupStateRequestData.PartitionData data, int snapshotEpoch) {
+ return new ShareGroupOffset(snapshotEpoch,
+ data.stateEpoch(),
+ data.leaderEpoch(),
+ data.startOffset(),
+ data.stateBatches().stream()
+ .map(PersisterStateBatch::from)
+ .collect(Collectors.toList()));
+ }
+
+ public Set<PersisterOffsetsStateBatch> stateBatchAsSet() {
+ return this.stateBatches().stream()
+ .map(PersisterOffsetsStateBatch::of)
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ }
+
+ public static class Builder {
+ private int snapshotEpoch;
+ private int stateEpoch;
+ private int leaderEpoch;
+ private long startOffset;
+ private List<? extends PersisterStateBatch> stateBatches;
+
+ public Builder setSnapshotEpoch(int snapshotEpoch) {
+ this.snapshotEpoch = snapshotEpoch;
+ return this;
+ }
+
+ public Builder setStateEpoch(int stateEpoch) {
+ this.stateEpoch = stateEpoch;
+ return this;
+ }
+
+ public Builder setLeaderEpoch(int leaderEpoch) {
+ this.leaderEpoch = leaderEpoch;
+ return this;
+ }
+
+ public Builder setStartOffset(long startOffset) {
+ this.startOffset = startOffset;
+ return this;
+ }
+
+ public Builder setStateBatches(List<? extends PersisterStateBatch> stateBatches) {
+ this.stateBatches = stateBatches;
+ return this;
+ }
+
+ public ShareGroupOffset build() {
+ return new ShareGroupOffset(snapshotEpoch, stateEpoch, leaderEpoch, startOffset, stateBatches);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ ShareGroupOffset that = (ShareGroupOffset) o;
+ return snapshotEpoch == that.snapshotEpoch && stateEpoch == that.stateEpoch && leaderEpoch == that.leaderEpoch && startOffset == that.startOffset && Objects.equals(stateBatches, that.stateBatches);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(snapshotEpoch, stateEpoch, leaderEpoch, startOffset, stateBatches);
+ }
Review Comment:
Can you add a toString()? These objects could end up in log messages
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Container class to represent data encapsulated in {@link ShareSnapshotValue} and {@link ShareUpdateValue}
+ * This class is effectively immutable (state batches is not modified out of context).
+ */
+public class ShareGroupOffset {
+ private final int snapshotEpoch;
+ private final int stateEpoch;
+ private final int leaderEpoch;
+ private final long startOffset;
+ private final List<? extends PersisterStateBatch> stateBatches;
+
+ private ShareGroupOffset(int snapshotEpoch,
+ int stateEpoch,
+ int leaderEpoch,
+ long startOffset,
+ List<? extends PersisterStateBatch> stateBatches) {
+ this.snapshotEpoch = snapshotEpoch;
+ this.stateEpoch = stateEpoch;
+ this.leaderEpoch = leaderEpoch;
+ this.startOffset = startOffset;
+ this.stateBatches = stateBatches;
+ }
+
+ public int snapshotEpoch() {
+ return snapshotEpoch;
+ }
+
+ public int stateEpoch() {
+ return stateEpoch;
+ }
+
+ public int leaderEpoch() {
+ return leaderEpoch;
+ }
+
+ public long startOffset() {
+ return startOffset;
+ }
+
+ public List<? extends PersisterStateBatch> stateBatches() {
+ return Collections.unmodifiableList(stateBatches);
+ }
+
+ private static PersisterStateBatch toPersisterStateBatch(ShareSnapshotValue.StateBatch stateBatch) {
+ return new PersisterStateBatch(stateBatch.firstOffset(), stateBatch.lastOffset(), stateBatch.deliveryState(), stateBatch.deliveryCount());
+ }
+
+ private static PersisterStateBatch toPersisterStateBatch(ShareUpdateValue.StateBatch stateBatch) {
+ return new PersisterStateBatch(stateBatch.firstOffset(), stateBatch.lastOffset(), stateBatch.deliveryState(), stateBatch.deliveryCount());
+ }
+
+ public static ShareGroupOffset fromRecord(ShareSnapshotValue record) {
+ return new ShareGroupOffset(record.snapshotEpoch(), record.stateEpoch(), record.leaderEpoch(), record.startOffset(), record.stateBatches().stream()
+ .map(ShareGroupOffset::toPersisterStateBatch).collect(Collectors.toList()));
+ }
+
+ public static ShareGroupOffset fromRecord(ShareUpdateValue record) {
+ return new ShareGroupOffset(record.snapshotEpoch(), -1, record.leaderEpoch(), record.startOffset(), record.stateBatches().stream()
+ .map(ShareGroupOffset::toPersisterStateBatch).collect(Collectors.toList()));
+ }
+
+ public static ShareGroupOffset fromRequest(WriteShareGroupStateRequestData.PartitionData data) {
+ return fromRequest(data, 0);
+ }
+
+ public static ShareGroupOffset fromRequest(WriteShareGroupStateRequestData.PartitionData data, int snapshotEpoch) {
+ return new ShareGroupOffset(snapshotEpoch,
+ data.stateEpoch(),
+ data.leaderEpoch(),
+ data.startOffset(),
+ data.stateBatches().stream()
+ .map(PersisterStateBatch::from)
+ .collect(Collectors.toList()));
+ }
+
+ public Set<PersisterOffsetsStateBatch> stateBatchAsSet() {
+ return this.stateBatches().stream()
+ .map(PersisterOffsetsStateBatch::of)
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ }
+
+ public static class Builder {
+ private int snapshotEpoch;
+ private int stateEpoch;
+ private int leaderEpoch;
+ private long startOffset;
Review Comment:
These will default to 0. Is that okay?
Another approach is to use the boxed types, then a null check in `build()`
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterOffsetsStateBatch.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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.Objects;
+
+/**
+ * This is a helper class which overrides the equals and hashcode
+ * methods to only focus on the first and last offset fields of the
+ * state batch. This is useful when combining batches.
+ */
+public class PersisterOffsetsStateBatch extends PersisterStateBatch {
Review Comment:
Using the type hierarchy for this is a bit unintuative and likely to cause surprises down the road. Imagine you have a `PersisterStateBatch` reference but it's actually a `PersisterOffsetsStateBatch` object and the behavior is different.
A better approach is to create a wrapper class that has the custom equalsTo and hashCode. In other words, let this class hold a PersisterStateBatch rather than extending that type.
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
Review Comment:
we need accessors for these values. I think a `TopicIdPartition topicIdPartition()` accessor would be useful too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730892020
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
+
+ /**
+ * Return the partition index for the given key.
+ *
+ * @param key - groupId:topicId:partitionId.
+ * @return The partition index.
+ */
+ int partitionFor(String key);
+
+ /**
+ * Return the configuration properties of the share-group state topic.
+ *
+ * @return Properties of the share-group state topic.
+ */
+ Properties shareGroupStateTopicConfigs();
+
+ /**
+ * Start the share coordinator
+ *
+ * @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
+ */
+ void startup(IntSupplier shareGroupTopicPartitionCount);
+
+ /**
+ * Stop the share coordinator
+ */
+ void shutdown();
+
+ /**
+ * Handle write share state call
+ * @param context - represents the incoming write request context
+ * @param request - actual RPC request object
+ * @return completable future comprizing of write RPC response data
Review Comment:
nit: `comprising`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730879998
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
Review Comment:
> Since the original code has been working and I did not author it, I will not be able to comment on the design.
That is fine, as the question I raised is unrelated to this PR. thanks for the discussion :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730874138
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
Review Comment:
So all this is existing code - it was initially part of GroupCoordinatorRecordSerde.java. Since share coordinator has extra records unrelated to group coordinator, common code was generalized to CoordinatorRecordSerde class and both group and share coordinator inherit from it.
Since the original code has been working and I did not author it, I will not be able to comment on the design.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727238710
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
+
+ public SharePartitionKey(String groupId, Uuid topicId, int partition) {
+ this.groupId = groupId;
+ this.topicId = topicId;
+ this.partition = partition;
+ }
+
+ public static class Builder {
+ private String groupId;
+ private Uuid topicId;
+ private int partition;
+
+ public Builder setGroupId(String groupId) {
+ this.groupId = groupId;
+ return this;
+ }
+
+ public Builder setTopicId(Uuid topicId) {
+ this.topicId = topicId;
+ return this;
+ }
+
+ public Builder setPartition(int partition) {
+ this.partition = partition;
+ return this;
+ }
+
+ public SharePartitionKey build() {
+ return new SharePartitionKey(groupId, topicId, partition);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (!(o instanceof SharePartitionKey)) return false;
+ SharePartitionKey that = (SharePartitionKey) o;
+ return partition == that.partition && Objects.equals(groupId, that.groupId) && Objects.equals(topicId, that.topicId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(groupId, topicId, partition);
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s:%s:%d", groupId, topicId, partition);
Review Comment:
We need share part keys `(groupId:topicId:partition)` as strings as well because that is mandated by some APIs like the FindCoordinator requests.
Keeping this format will allow to use `SharePartitionKey` interchangeably.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727572775
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
Review Comment:
> I think a TopicIdPartition topicIdPartition() accessor would be useful too.
Can't do that. I cannot construct a TopicIdPartition object to return since we do not know the topic name and that is necessary to create TopicIdPartition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727247854
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
+
+ public SharePartitionKey(String groupId, Uuid topicId, int partition) {
+ this.groupId = groupId;
+ this.topicId = topicId;
+ this.partition = partition;
+ }
+
+ public static class Builder {
Review Comment:
Can't use `(String, TopicIdPartition)`. All coordinator keys use topic specified as Uuid.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727247854
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
+
+ public SharePartitionKey(String groupId, Uuid topicId, int partition) {
+ this.groupId = groupId;
+ this.topicId = topicId;
+ this.partition = partition;
+ }
+
+ public static class Builder {
Review Comment:
Can't use `(String, TopicIdPartition)`. Share coordinator keys use topic specified as Uuid.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727598130
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterOffsetsStateBatch.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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.Objects;
+
+/**
+ * This is a helper class which overrides the equals and hashcode
+ * methods to only focus on the first and last offset fields of the
+ * state batch. This is useful when combining batches.
+ */
+public class PersisterOffsetsStateBatch extends PersisterStateBatch {
Review Comment:
will use the delegate
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730816423
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
Review Comment:
> Yes, we can keep this 0 - not necessary to use constant.
yes, that is what I expected :)
> The constant is needed in apiMessageValueFor impl in ShareCoordinatorRecordSerde.
The CoordinatorRecordSerde.deserialize method reads the key version and supplies the same while fetching the ApiMessage object to read the data into
thanks for the explanation!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730841225
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
Review Comment:
> For each json file - there is an autogenerated java file which has a read method which has that validation.
IIRC, the auto-generated java file does not check the version range. the unknown (newer) message may have new fields and the "old" reader can see error if it is lucky. otherwise, it may get incorrect parsed values.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1723957737
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordSerde.java:
##########
@@ -53,90 +49,11 @@
import org.apache.kafka.coordinator.group.generated.ShareGroupTargetAssignmentMemberValue;
import org.apache.kafka.coordinator.group.generated.ShareGroupTargetAssignmentMetadataKey;
import org.apache.kafka.coordinator.group.generated.ShareGroupTargetAssignmentMetadataValue;
-import org.apache.kafka.server.common.ApiMessageAndVersion;
-import java.nio.BufferUnderflowException;
-import java.nio.ByteBuffer;
-
-/**
- * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
- * <pre>
- * record_key = [record_type key_message]
- * record_value = [value_version value_message]
- *
- * record_type : The record type is currently define as the version of the key
- * {@link ApiMessageAndVersion} object.
- * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
- * value_version : The value version is currently define as the version of the value
- * {@link ApiMessageAndVersion} object.
- * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
- * </pre>
- */
@SuppressWarnings({ "ClassDataAbstractionCoupling", "CyclomaticComplexity" })
Review Comment:
Yes
```
[ant:checkstyle] [ERROR] /home/smjn/ws/kafka/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordSerde.java:54:1: Class Data Abstraction Coupling is 31 (max allowed is 25)
...
[ClassDataAbstractionCoupling]
[ant:checkstyle] [ERROR] /home/smjn/ws/kafka/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordSerde.java:55:5: Cyclomatic Complexity is 17 (max allowed is 16). [CyclomaticComplexity]
[ant:checkstyle] [ERROR] /home/smjn/ws/kafka/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordSerde.java:94:5: Cyclomatic Complexity is 17 (max allowed is 16). [CyclomaticComplexity]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1723973298
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
+
+ /**
+ * Return the partition index for the given key.
+ *
+ * @param key key - groupId:topicId:partitionId.
+ * @return The partition index.
+ */
+ int partitionFor(String key);
+
+ /**
+ * Return the configuration properties of the share-group state topic.
+ *
+ * @return Properties of the share-group state topic.
+ */
+ Properties shareGroupStateTopicConfigs();
+
+ /**
+ * Start the share coordinator
+ *
+ * @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
+ */
+ void startup(IntSupplier shareGroupTopicPartitionCount);
+
+ /**
+ * Stop the share coordinator
+ */
+ void shutdown();
+
+ /**
+ * Handle write share state call
+ * @param context - represents the incoming write request context
Review Comment:
code exists with and without `-` in the kafka codebase. The - has been used consistently in the interface in question. Refactoring entire kafka codebase to be consistent might be beyond the scope of this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1722862341
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+@InterfaceStability.Evolving
Review Comment:
I don't think we need to mark an internal interface as Evolving, even if we do expect to change it. All of the consumers of the interface are also in AK and can be changed consistently and atomically without incompatibility problems.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Container class to represent data encapsulated in @link ShareSnapshotValue and @link ShareUpdateValue
+ * This class is effectively immutable (state batches is not modified out of context).
+ */
+public class ShareGroupOffset {
+ private final int snapshotEpoch;
+ private final int stateEpoch;
+ private final int leaderEpoch;
+ private final long startOffset;
+ private final List<? extends PersisterStateBatch> stateBatches;
+
+ public ShareGroupOffset(int snapshotEpoch,
+ int stateEpoch,
+ int leaderEpoch,
+ long startOffset,
+ List<? extends PersisterStateBatch> stateBatches) {
+ this.snapshotEpoch = snapshotEpoch;
+ this.stateEpoch = stateEpoch;
+ this.leaderEpoch = leaderEpoch;
+ this.startOffset = startOffset;
+ this.stateBatches = stateBatches;
+ }
+
+ public int snapshotEpoch() {
+ return snapshotEpoch;
+ }
+
+ public int stateEpoch() {
+ return stateEpoch;
+ }
+
+ public int leaderEpoch() {
+ return leaderEpoch;
+ }
+
+ public long startOffset() {
+ return startOffset;
+ }
+
+ public List<? extends PersisterStateBatch> stateBatches() {
+ return stateBatches;
Review Comment:
Can you enforce the immutability by returning an unmodifiable list in this case? Seems quite simple to wrap it and avoid embarrassing bugs later.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordSerde.java:
##########
@@ -155,6 +72,8 @@ private ApiMessage apiMessageKeyFor(short recordType) {
return new ConsumerGroupTargetAssignmentMemberKey();
case 8:
return new ConsumerGroupCurrentMemberAssignmentKey();
+ // the following share group api messages are handled by
Review Comment:
I don't see the need for this comment. These records are the GC record types. There are consumer group records, share group records, and there will be streams group records once KIP-1071 is approved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730727174
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
Review Comment:
@chia7712
Not sure about this - with this change a lot of tests in `GroupCoordinatorRecordSerdeTest` will fail.
IDK if those tests are deliberate.
They pass differing values for versions of keys and values (`testDeserialize`). With the above change they will pass if the versions are modified as per `apiMessageFor(recordType)` implementation. That would need further analysis of `GroupCoordinatorRecordHelpers`.
I would be hesitant to make this change unless the original author @dajac could provide some input.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730728078
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
Review Comment:
Related to https://github.com/apache/kafka/pull/16921#discussion_r1730727174
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730825016
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
Review Comment:
Yes, that validation code is added as part of auto-generated code.
For each json file - there is an autogenerated java file which has a read method which has that validation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727377885
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
+
+ public SharePartitionKey(String groupId, Uuid topicId, int partition) {
+ this.groupId = groupId;
+ this.topicId = topicId;
+ this.partition = partition;
+ }
+
+ public static class Builder {
+ private String groupId;
+ private Uuid topicId;
+ private int partition;
+
+ public Builder setGroupId(String groupId) {
+ this.groupId = groupId;
+ return this;
+ }
+
+ public Builder setTopicId(Uuid topicId) {
+ this.topicId = topicId;
+ return this;
+ }
+
+ public Builder setPartition(int partition) {
+ this.partition = partition;
+ return this;
+ }
+
+ public SharePartitionKey build() {
+ return new SharePartitionKey(groupId, topicId, partition);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (!(o instanceof SharePartitionKey)) return false;
+ SharePartitionKey that = (SharePartitionKey) o;
+ return partition == that.partition && Objects.equals(groupId, that.groupId) && Objects.equals(topicId, that.topicId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(groupId, topicId, partition);
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s:%s:%d", groupId, topicId, partition);
Review Comment:
committed in the subsequent revision, plz check once again
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727247854
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
+
+ public SharePartitionKey(String groupId, Uuid topicId, int partition) {
+ this.groupId = groupId;
+ this.topicId = topicId;
+ this.partition = partition;
+ }
+
+ public static class Builder {
Review Comment:
Can't use `(String, TopicIdPartition)`. Share coordinator keys use topic specified as Uuid.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730810918
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
Review Comment:
@chia7712
Yes, we can keep this 0 - not necessary to use constant.
The constant is needed in `apiMessageValueFor` impl in ShareCoordinatorRecordSerde.
The `CoordinatorRecordSerde.deserialize` method reads the key version and supplies the same while fetching the ApiMessage object to read the data into
```
final ApiMessage valueMessage = apiMessageValueFor(recordType); // recordType is keyversion -used to fetch correct object to read data into
final short valueVersion = readVersion(valueBuffer, "value");
readMessage(valueMessage, valueBuffer, valueVersion, "value"); // data is read into the messageObject
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730810918
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
Review Comment:
@chia7712
Yes, we can keep this 0 - not necessary to use constant.
The constant is needed in `apiMessageValueFor` impl in ShareCoordinatorRecordSerde.
The `CoordinatorRecordSerde.deserialize` method reads the key version and supplies the same while fetching the ApiMessage object to read the data into
```
final ApiMessage valueMessage = apiMessageValueFor(recordType); // recordType is keyversion
final short valueVersion = readVersion(valueBuffer, "value");
readMessage(valueMessage, valueBuffer, valueVersion, "value"); // data is read into the messageObject
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730849021
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
Review Comment:
The version checks are for the data fields and the "validVersions" record field.
Additionally, there are 2 methods hightestSupportedVersion and lowestSupportedVersion which specify the range
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1726384651
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
+
+ /**
+ * Return the partition index for the given key.
+ *
+ * @param key key - groupId:topicId:partitionId.
+ * @return The partition index.
+ */
+ int partitionFor(String key);
+
+ /**
+ * Return the configuration properties of the share-group state topic.
+ *
+ * @return Properties of the share-group state topic.
+ */
+ Properties shareGroupStateTopicConfigs();
+
+ /**
+ * Start the share coordinator
+ *
+ * @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
+ */
+ void startup(IntSupplier shareGroupTopicPartitionCount);
+
+ /**
+ * Stop the share coordinator
+ */
+ void shutdown();
+
+ /**
+ * Handle write share state call
+ * @param context - represents the incoming write request context
Review Comment:
Following
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1723968864
##########
share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorRecordSerdeTest.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorLoader;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotKey;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateKey;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ShareCoordinatorRecordSerdeTest {
+ private ShareCoordinatorRecordSerde serde;
+
+ @BeforeEach
+ public void setup() {
+ serde = new ShareCoordinatorRecordSerde();
+ }
+
+ @Test
+ public void testSerializeKey() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+
+ assertArrayEquals(
+ MessageUtil.toVersionPrefixedBytes(record.key().version(), record.key().message()),
+ serde.serializeKey(record)
+ );
+ }
+
+ @Test
+ public void testSerializeValue() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+
+ assertArrayEquals(
+ MessageUtil.toVersionPrefixedBytes(record.value().version(), record.value().message()),
+ serde.serializeValue(record)
+ );
+ }
+
+ @Test
+ public void testSerializeNullValue() {
+ CoordinatorRecord record = new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("group")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ ),
+ null
+ );
+
+ assertNull(serde.serializeValue(record));
+ }
+
+ @Test
+ public void testDeserialize() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+ ApiMessageAndVersion key = record.key();
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ApiMessageAndVersion value = record.value();
+ ByteBuffer valueBuffer = MessageUtil.toVersionPrefixedByteBuffer(value.version(), value.message());
+
+ CoordinatorRecord desRecord = serde.deserialize(keyBuffer, valueBuffer);
+ assertEquals(key, desRecord.key());
+ assertEquals(value, desRecord.value());
+ }
+
+ @Test
+ public void testDeserializeWithTombstoneForValue() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("groupId")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ CoordinatorRecord record = serde.deserialize(keyBuffer, null);
+ assertEquals(key, record.key());
+ assertNull(record.value());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidRecordType() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(64);
+ keyBuffer.putShort((short) 255);
+ keyBuffer.rewind();
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(64);
+
+ CoordinatorLoader.UnknownRecordTypeException ex =
+ assertThrows(CoordinatorLoader.UnknownRecordTypeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals((short) 255, ex.unknownType());
+ }
+
+ @Test
+ public void testDeserializeWithKeyEmptyBuffer() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(0);
+ ByteBuffer valueBuffer = ByteBuffer.allocate(64);
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals("Could not read version from key's buffer.", ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithValueEmptyBuffer() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("foo")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(0);
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals("Could not read version from value's buffer.", ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidKeyBytes() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(2);
+ keyBuffer.putShort((short) 0);
+ keyBuffer.rewind();
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(2);
+ valueBuffer.putShort((short) 0);
+ valueBuffer.rewind();
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertTrue(ex.getMessage().startsWith("Could not read record with version 0 from key's buffer due to"),
+ ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidValueBytes() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("foo")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(2);
+ valueBuffer.putShort((short) 0);
+ valueBuffer.rewind();
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertTrue(ex.getMessage().startsWith("Could not read record with version 0 from value's buffer due to"),
+ ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeAllRecordTypes() {
+ roundTrip((short) 0, new ShareSnapshotKey(), new ShareSnapshotValue());
Review Comment:
No, it does not. This test is just checking the deserialise method. The message is converted to bytes and then deserialized to check if it is same as original message.
There is no validation on the data or version in this process.
In the actual impl we rely only on the version field of the incoming byte buffer to determine what type of record we want to form.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727201011
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
+
+ public SharePartitionKey(String groupId, Uuid topicId, int partition) {
+ this.groupId = groupId;
+ this.topicId = topicId;
+ this.partition = partition;
+ }
+
+ public static class Builder {
+ private String groupId;
+ private Uuid topicId;
+ private int partition;
+
+ public Builder setGroupId(String groupId) {
+ this.groupId = groupId;
+ return this;
+ }
+
+ public Builder setTopicId(Uuid topicId) {
+ this.topicId = topicId;
+ return this;
+ }
+
+ public Builder setPartition(int partition) {
+ this.partition = partition;
+ return this;
+ }
+
+ public SharePartitionKey build() {
+ return new SharePartitionKey(groupId, topicId, partition);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (!(o instanceof SharePartitionKey)) return false;
+ SharePartitionKey that = (SharePartitionKey) o;
+ return partition == that.partition && Objects.equals(groupId, that.groupId) && Objects.equals(topicId, that.topicId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(groupId, topicId, partition);
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s:%s:%d", groupId, topicId, partition);
Review Comment:
Is there a reason we don't want the usual "toString" format here?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Container class to represent data encapsulated in @link ShareSnapshotValue and @link ShareUpdateValue
+ * This class is effectively immutable (state batches is not modified out of context).
+ */
+public class ShareGroupOffset {
+ private final int snapshotEpoch;
+ private final int stateEpoch;
+ private final int leaderEpoch;
+ private final long startOffset;
+ private final List<? extends PersisterStateBatch> stateBatches;
+
+ public ShareGroupOffset(int snapshotEpoch,
+ int stateEpoch,
+ int leaderEpoch,
+ long startOffset,
Review Comment:
Methods/constructors with many sequential arguments of the same type are prone to swapped argument types of bugs. I would suggest making this constructor private and forcing callers to use the builder.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterOffsetsStateBatch.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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.Objects;
+
+/**
+ * This is a helper class which overrides the equals and hashcode
+ * methods to only focus on the first and last offset fields of the
+ * state batch. This is useful when combining batches.
+ */
+public class PersisterOffsetsStateBatch extends PersisterStateBatch {
+ public PersisterOffsetsStateBatch(long firstOffset, long lastOffset, byte deliveryState, short deliveryCount) {
+ super(firstOffset, lastOffset, deliveryState, deliveryCount);
+ }
+
+ public static PersisterOffsetsStateBatch of(ShareSnapshotValue.StateBatch batch) {
+ return new PersisterOffsetsStateBatch(batch.firstOffset(), batch.lastOffset(), batch.deliveryState(), batch.deliveryCount());
+ }
+
+ public static PersisterOffsetsStateBatch of(ShareUpdateValue.StateBatch batch) {
+ return new PersisterOffsetsStateBatch(batch.firstOffset(), batch.lastOffset(), batch.deliveryState(), batch.deliveryCount());
+ }
+
+ public static PersisterOffsetsStateBatch of(PersisterStateBatch batch) {
+ return new PersisterOffsetsStateBatch(batch.firstOffset(), batch.lastOffset(), batch.deliveryState(), batch.deliveryCount());
+ }
+
+ public static PersisterOffsetsStateBatch of(WriteShareGroupStateRequestData.StateBatch batch) {
+ return new PersisterOffsetsStateBatch(batch.firstOffset(), batch.lastOffset(), batch.deliveryState(), batch.deliveryCount());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof PersisterOffsetsStateBatch)) {
+ return false;
+ }
+ PersisterOffsetsStateBatch that = (PersisterOffsetsStateBatch) o;
+ return this.firstOffset() == that.firstOffset() && this.lastOffset() == that.lastOffset();
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(firstOffset(), lastOffset());
+ }
+}
Review Comment:
nit: missing newline
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordSerde.java:
##########
@@ -53,90 +49,11 @@
import org.apache.kafka.coordinator.group.generated.ShareGroupTargetAssignmentMemberValue;
import org.apache.kafka.coordinator.group.generated.ShareGroupTargetAssignmentMetadataKey;
import org.apache.kafka.coordinator.group.generated.ShareGroupTargetAssignmentMetadataValue;
-import org.apache.kafka.server.common.ApiMessageAndVersion;
-import java.nio.BufferUnderflowException;
-import java.nio.ByteBuffer;
-
-/**
- * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
- * <pre>
- * record_key = [record_type key_message]
- * record_value = [value_version value_message]
- *
- * record_type : The record type is currently define as the version of the key
- * {@link ApiMessageAndVersion} object.
- * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
- * value_version : The value version is currently define as the version of the value
- * {@link ApiMessageAndVersion} object.
- * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
- * </pre>
- */
@SuppressWarnings({ "ClassDataAbstractionCoupling", "CyclomaticComplexity" })
Review Comment:
There is a better way to suppress checkstyle warnings. Look at "checkstyle/suppressions.xml"
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
+
+ public SharePartitionKey(String groupId, Uuid topicId, int partition) {
+ this.groupId = groupId;
+ this.topicId = topicId;
+ this.partition = partition;
+ }
+
+ public static class Builder {
Review Comment:
Why do we need a builder for this class? Three arguments seems fine to deal with, and I doubt it will evolve over time. How about a since private constructor, and two public static factories: one (String, Uuid, int) and one (String, TopicIdPartition)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730727174
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
Review Comment:
@chia7712
Not sure about this - with this change a lot of tests in `GroupCoordinatorRecordSerdeTest` will fail.
IDK if those tests are deliberate.
The tests use differing arguments for versions of keys and values (`testDeserialize`). With the above change they will pass if the versions are modified as per `apiMessageFor(recordType)` implementation in GroupCoordinatorRecordSerde. That would need further analysis of `GroupCoordinatorRecordHelpers`.
I would be hesitant to make this change unless the original author @dajac could provide some input.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730810918
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
Review Comment:
Yes, we can keep this 0 - not necessary to use constant.
The constant is needed in `apiMessageValueFor` impl in ShareCoordinatorRecordSerde.
The `CoordinatorRecordSerde.deserialize` method reads the key version and supplies the same while fetching the ApiMessage object to read the data into
```
final ApiMessage valueMessage = apiMessageValueFor(recordType); // recordType is keyversion
final short valueVersion = readVersion(valueBuffer, "value");
readMessage(valueMessage, valueBuffer, valueVersion, "value"); // data is read into the messageObject
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730849021
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
Review Comment:
The version checks are for the fields and the "validVersions"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730790598
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
Review Comment:
ok, I gets confused by the `SHARE_UPDATE_RECORD_VALUE_VERSION`. Let me re-comment it again.
`final ApiMessage valueMessage = apiMessageValueFor(recordType);` is correct, as the type of value is associated to key, so they share the same `recordType`. that is good.
The code which confuses me is shown below. `SHARE_UPDATE_RECORD_VALUE_VERSION` is `1` and `ShareUpdateValue`'s version is (0, 0). I assume the version added to `ApiMessageAndVersion` is the protocol version. Please correct me if I'm wrong
```java
new ApiMessageAndVersion(new ShareUpdateValue()
.setSnapshotEpoch(offsetData.snapshotEpoch())
.setLeaderEpoch(offsetData.leaderEpoch())
.setStartOffset(offsetData.startOffset())
.setStateBatches(offsetData.stateBatches().stream()
.map(batch -> new ShareUpdateValue.StateBatch()
.setFirstOffset(batch.firstOffset())
.setLastOffset(batch.lastOffset())
.setDeliveryCount(batch.deliveryCount())
.setDeliveryState(batch.deliveryState()))
.collect(Collectors.toList())),
ShareCoordinator.SHARE_UPDATE_RECORD_VALUE_VERSION)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727577200
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareGroupOffset.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Container class to represent data encapsulated in {@link ShareSnapshotValue} and {@link ShareUpdateValue}
+ * This class is effectively immutable (state batches is not modified out of context).
+ */
+public class ShareGroupOffset {
+ private final int snapshotEpoch;
+ private final int stateEpoch;
+ private final int leaderEpoch;
+ private final long startOffset;
+ private final List<? extends PersisterStateBatch> stateBatches;
+
+ private ShareGroupOffset(int snapshotEpoch,
+ int stateEpoch,
+ int leaderEpoch,
+ long startOffset,
+ List<? extends PersisterStateBatch> stateBatches) {
+ this.snapshotEpoch = snapshotEpoch;
+ this.stateEpoch = stateEpoch;
+ this.leaderEpoch = leaderEpoch;
+ this.startOffset = startOffset;
+ this.stateBatches = stateBatches;
+ }
+
+ public int snapshotEpoch() {
+ return snapshotEpoch;
+ }
+
+ public int stateEpoch() {
+ return stateEpoch;
+ }
+
+ public int leaderEpoch() {
+ return leaderEpoch;
+ }
+
+ public long startOffset() {
+ return startOffset;
+ }
+
+ public List<? extends PersisterStateBatch> stateBatches() {
+ return Collections.unmodifiableList(stateBatches);
+ }
+
+ private static PersisterStateBatch toPersisterStateBatch(ShareSnapshotValue.StateBatch stateBatch) {
+ return new PersisterStateBatch(stateBatch.firstOffset(), stateBatch.lastOffset(), stateBatch.deliveryState(), stateBatch.deliveryCount());
+ }
+
+ private static PersisterStateBatch toPersisterStateBatch(ShareUpdateValue.StateBatch stateBatch) {
+ return new PersisterStateBatch(stateBatch.firstOffset(), stateBatch.lastOffset(), stateBatch.deliveryState(), stateBatch.deliveryCount());
+ }
+
+ public static ShareGroupOffset fromRecord(ShareSnapshotValue record) {
+ return new ShareGroupOffset(record.snapshotEpoch(), record.stateEpoch(), record.leaderEpoch(), record.startOffset(), record.stateBatches().stream()
+ .map(ShareGroupOffset::toPersisterStateBatch).collect(Collectors.toList()));
+ }
+
+ public static ShareGroupOffset fromRecord(ShareUpdateValue record) {
+ return new ShareGroupOffset(record.snapshotEpoch(), -1, record.leaderEpoch(), record.startOffset(), record.stateBatches().stream()
+ .map(ShareGroupOffset::toPersisterStateBatch).collect(Collectors.toList()));
+ }
+
+ public static ShareGroupOffset fromRequest(WriteShareGroupStateRequestData.PartitionData data) {
+ return fromRequest(data, 0);
+ }
+
+ public static ShareGroupOffset fromRequest(WriteShareGroupStateRequestData.PartitionData data, int snapshotEpoch) {
+ return new ShareGroupOffset(snapshotEpoch,
+ data.stateEpoch(),
+ data.leaderEpoch(),
+ data.startOffset(),
+ data.stateBatches().stream()
+ .map(PersisterStateBatch::from)
+ .collect(Collectors.toList()));
+ }
+
+ public Set<PersisterOffsetsStateBatch> stateBatchAsSet() {
+ return this.stateBatches().stream()
+ .map(PersisterOffsetsStateBatch::of)
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ }
+
+ public static class Builder {
+ private int snapshotEpoch;
+ private int stateEpoch;
+ private int leaderEpoch;
+ private long startOffset;
Review Comment:
Yes, zero is valid default value for these fields.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1723969457
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ public final String groupId;
+ public final Uuid topicId;
+ public final int partition;
Review Comment:
no, but since the class is immutable - we can used pub fields instead of getter methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730811753
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
Review Comment:
out of curiosity. Do we have validation of the version range? or that is unnecessary to add such validation for those messages?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1727572775
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
Review Comment:
> we need accessors for these values.
Will add - but they are not utilized anywhere
> I think a TopicIdPartition topicIdPartition() accessor would be useful too.
Can't do that. I cannot construct a TopicIdPartition object to return since we do not know the topic name and that is necessary to create TopicIdPartition.
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ private final String groupId;
+ private final Uuid topicId;
+ private final int partition;
Review Comment:
> we need accessors for these values.
Will add - but they are not utilized anywhere
> I think a TopicIdPartition topicIdPartition() accessor would be useful too.
Can't do that. I cannot construct a TopicIdPartition object to return since we do not know the topic name and that is necessary to create TopicIdPartition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730849021
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
Review Comment:
The version checks are for the fields and the "validVersions"
Additionally, there are 2 methods hightestSupportedVersion and lowestSupportedVersion which specify the range
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1725687001
##########
share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorRecordSerdeTest.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorLoader;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotKey;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateKey;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ShareCoordinatorRecordSerdeTest {
+ private ShareCoordinatorRecordSerde serde;
+
+ @BeforeEach
+ public void setup() {
+ serde = new ShareCoordinatorRecordSerde();
+ }
+
+ @Test
+ public void testSerializeKey() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+
+ assertArrayEquals(
+ MessageUtil.toVersionPrefixedBytes(record.key().version(), record.key().message()),
+ serde.serializeKey(record)
+ );
+ }
+
+ @Test
+ public void testSerializeValue() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+
+ assertArrayEquals(
+ MessageUtil.toVersionPrefixedBytes(record.value().version(), record.value().message()),
+ serde.serializeValue(record)
+ );
+ }
+
+ @Test
+ public void testSerializeNullValue() {
+ CoordinatorRecord record = new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("group")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ ),
+ null
+ );
+
+ assertNull(serde.serializeValue(record));
+ }
+
+ @Test
+ public void testDeserialize() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+ ApiMessageAndVersion key = record.key();
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ApiMessageAndVersion value = record.value();
+ ByteBuffer valueBuffer = MessageUtil.toVersionPrefixedByteBuffer(value.version(), value.message());
+
+ CoordinatorRecord desRecord = serde.deserialize(keyBuffer, valueBuffer);
+ assertEquals(key, desRecord.key());
+ assertEquals(value, desRecord.value());
+ }
+
+ @Test
+ public void testDeserializeWithTombstoneForValue() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("groupId")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ CoordinatorRecord record = serde.deserialize(keyBuffer, null);
+ assertEquals(key, record.key());
+ assertNull(record.value());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidRecordType() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(64);
+ keyBuffer.putShort((short) 255);
+ keyBuffer.rewind();
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(64);
+
+ CoordinatorLoader.UnknownRecordTypeException ex =
+ assertThrows(CoordinatorLoader.UnknownRecordTypeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals((short) 255, ex.unknownType());
+ }
+
+ @Test
+ public void testDeserializeWithKeyEmptyBuffer() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(0);
+ ByteBuffer valueBuffer = ByteBuffer.allocate(64);
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals("Could not read version from key's buffer.", ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithValueEmptyBuffer() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("foo")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(0);
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals("Could not read version from value's buffer.", ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidKeyBytes() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(2);
+ keyBuffer.putShort((short) 0);
+ keyBuffer.rewind();
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(2);
+ valueBuffer.putShort((short) 0);
+ valueBuffer.rewind();
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertTrue(ex.getMessage().startsWith("Could not read record with version 0 from key's buffer due to"),
+ ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidValueBytes() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("foo")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(2);
+ valueBuffer.putShort((short) 0);
+ valueBuffer.rewind();
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertTrue(ex.getMessage().startsWith("Could not read record with version 0 from value's buffer due to"),
+ ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeAllRecordTypes() {
+ roundTrip((short) 0, new ShareSnapshotKey(), new ShareSnapshotValue());
Review Comment:
I think this test does check about the keys and passed number identifier, but not the values, correct? I am not sure what we can test with passed value message in this test. You might want to write the comment on top for clarification.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1725688735
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
+
+ /**
+ * Return the partition index for the given key.
+ *
+ * @param key key - groupId:topicId:partitionId.
+ * @return The partition index.
+ */
+ int partitionFor(String key);
+
+ /**
+ * Return the configuration properties of the share-group state topic.
+ *
+ * @return Properties of the share-group state topic.
+ */
+ Properties shareGroupStateTopicConfigs();
+
+ /**
+ * Start the share coordinator
+ *
+ * @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
+ */
+ void startup(IntSupplier shareGroupTopicPartitionCount);
+
+ /**
+ * Stop the share coordinator
+ */
+ void shutdown();
+
+ /**
+ * Handle write share state call
+ * @param context - represents the incoming write request context
Review Comment:
Sure, shall we just maintain consistency in this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1723912831
##########
checkstyle/import-control-share-coordinator.xml:
##########
@@ -30,11 +30,18 @@
<!-- anyone can use public classes -->
<subpackage name="coordinator">
<subpackage name="share">
+ <allow pkg="org.apache.kafka.common" />
+ <allow pkg="org.apache.kafka.common.annotation" />
<allow pkg="org.apache.kafka.common.config" />
<allow pkg="org.apache.kafka.common.message" />
+ <allow pkg="org.apache.kafka.common.protocol" />
<allow pkg="org.apache.kafka.common.requests" />
<allow pkg="org.apache.kafka.coordinator.common" />
+ <allow pkg="org.apache.kafka.coordinator.share.generated" />
<allow pkg="org.apache.kafka.image" />
+ <allow pkg="org.apache.kafka.server.common" />
+ <allow pkg="org.apache.kafka.server.group.share" />
+ <allow pkg="org.junit.jupiter.api" />
Review Comment:
You might want to consider moving it under `<!-- common library dependencies -->`
##########
share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorRecordSerdeTest.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorLoader;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotKey;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateKey;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ShareCoordinatorRecordSerdeTest {
+ private ShareCoordinatorRecordSerde serde;
+
+ @BeforeEach
+ public void setup() {
+ serde = new ShareCoordinatorRecordSerde();
+ }
+
+ @Test
+ public void testSerializeKey() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+
+ assertArrayEquals(
+ MessageUtil.toVersionPrefixedBytes(record.key().version(), record.key().message()),
+ serde.serializeKey(record)
+ );
+ }
+
+ @Test
+ public void testSerializeValue() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+
+ assertArrayEquals(
+ MessageUtil.toVersionPrefixedBytes(record.value().version(), record.value().message()),
+ serde.serializeValue(record)
+ );
+ }
+
+ @Test
+ public void testSerializeNullValue() {
+ CoordinatorRecord record = new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("group")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ ),
+ null
+ );
+
+ assertNull(serde.serializeValue(record));
+ }
+
+ @Test
+ public void testDeserialize() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+ ApiMessageAndVersion key = record.key();
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ApiMessageAndVersion value = record.value();
+ ByteBuffer valueBuffer = MessageUtil.toVersionPrefixedByteBuffer(value.version(), value.message());
+
+ CoordinatorRecord desRecord = serde.deserialize(keyBuffer, valueBuffer);
+ assertEquals(key, desRecord.key());
+ assertEquals(value, desRecord.value());
+ }
+
+ @Test
+ public void testDeserializeWithTombstoneForValue() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("groupId")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ CoordinatorRecord record = serde.deserialize(keyBuffer, null);
+ assertEquals(key, record.key());
+ assertNull(record.value());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidRecordType() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(64);
+ keyBuffer.putShort((short) 255);
+ keyBuffer.rewind();
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(64);
+
+ CoordinatorLoader.UnknownRecordTypeException ex =
+ assertThrows(CoordinatorLoader.UnknownRecordTypeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals((short) 255, ex.unknownType());
+ }
+
+ @Test
+ public void testDeserializeWithKeyEmptyBuffer() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(0);
+ ByteBuffer valueBuffer = ByteBuffer.allocate(64);
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals("Could not read version from key's buffer.", ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithValueEmptyBuffer() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("foo")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(0);
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals("Could not read version from value's buffer.", ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidKeyBytes() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(2);
+ keyBuffer.putShort((short) 0);
+ keyBuffer.rewind();
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(2);
+ valueBuffer.putShort((short) 0);
+ valueBuffer.rewind();
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertTrue(ex.getMessage().startsWith("Could not read record with version 0 from key's buffer due to"),
+ ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidValueBytes() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("foo")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(2);
+ valueBuffer.putShort((short) 0);
+ valueBuffer.rewind();
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertTrue(ex.getMessage().startsWith("Could not read record with version 0 from value's buffer due to"),
+ ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeAllRecordTypes() {
+ roundTrip((short) 0, new ShareSnapshotKey(), new ShareSnapshotValue());
Review Comment:
Does this test ensure that value passed is also verified against the key i.e. if I write mismatched key val `roundTrip((short) 0, new ShareSnapshotKey(), new ShareUpdateValue());` then will it fail?
##########
checkstyle/import-control-share-coordinator.xml:
##########
@@ -30,11 +30,18 @@
<!-- anyone can use public classes -->
<subpackage name="coordinator">
<subpackage name="share">
+ <allow pkg="org.apache.kafka.common" />
Review Comment:
It might be better to use the way other import controls are structured generally with
```
<!-- anyone can use public classes -->
<allow pkg="org.apache.kafka.common" exact-match="true" />
...
```
```
<!-- protocol, records and request/response utilities -->
<allow pkg="org.apache.kafka.clients" />
<allow pkg="org.apache.kafka.common.message" />
<allow pkg="org.apache.kafka.common.protocol" />
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordSerde.java:
##########
@@ -53,90 +49,11 @@
import org.apache.kafka.coordinator.group.generated.ShareGroupTargetAssignmentMemberValue;
import org.apache.kafka.coordinator.group.generated.ShareGroupTargetAssignmentMetadataKey;
import org.apache.kafka.coordinator.group.generated.ShareGroupTargetAssignmentMetadataValue;
-import org.apache.kafka.server.common.ApiMessageAndVersion;
-import java.nio.BufferUnderflowException;
-import java.nio.ByteBuffer;
-
-/**
- * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
- * <pre>
- * record_key = [record_type key_message]
- * record_value = [value_version value_message]
- *
- * record_type : The record type is currently define as the version of the key
- * {@link ApiMessageAndVersion} object.
- * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
- * value_version : The value version is currently define as the version of the value
- * {@link ApiMessageAndVersion} object.
- * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
- * </pre>
- */
@SuppressWarnings({ "ClassDataAbstractionCoupling", "CyclomaticComplexity" })
Review Comment:
Do we still need these warnings after refactoring?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
+
+ /**
+ * Return the partition index for the given key.
+ *
+ * @param key key - groupId:topicId:partitionId.
+ * @return The partition index.
+ */
+ int partitionFor(String key);
+
+ /**
+ * Return the configuration properties of the share-group state topic.
+ *
+ * @return Properties of the share-group state topic.
+ */
+ Properties shareGroupStateTopicConfigs();
+
+ /**
+ * Start the share coordinator
+ *
+ * @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
+ */
+ void startup(IntSupplier shareGroupTopicPartitionCount);
+
+ /**
+ * Stop the share coordinator
+ */
+ void shutdown();
+
+ /**
+ * Handle write share state call
+ * @param context - represents the incoming write request context
+ * @param request - actual RPC request object
+ * @return completable future comprizing of write RPC response data
+ */
+ CompletableFuture<WriteShareGroupStateResponseData> writeState(RequestContext context, WriteShareGroupStateRequestData request);
+
+
+ /**
+ * Handle read share state call
+ * @param context - represents the incoming read request context
+ * @param request - actual RPC request object
+ * @return completable future comprising of write RPC response data
Review Comment:
```suggestion
* @return completable future comprising of read RPC response data
```
##########
server-common/src/main/java/org/apache/kafka/server/group/share/SharePartitionKey.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.server.group.share;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Common immutable share partition key class. This class is
+ * placed in server-common so that it can be freely used across
+ * various modules.
+ */
+public class SharePartitionKey {
+ public final String groupId;
+ public final Uuid topicId;
+ public final int partition;
Review Comment:
Are they required to be public?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
+
+ /**
+ * Return the partition index for the given key.
+ *
+ * @param key key - groupId:topicId:partitionId.
+ * @return The partition index.
+ */
+ int partitionFor(String key);
+
+ /**
+ * Return the configuration properties of the share-group state topic.
+ *
+ * @return Properties of the share-group state topic.
+ */
+ Properties shareGroupStateTopicConfigs();
+
+ /**
+ * Start the share coordinator
+ *
+ * @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
+ */
+ void startup(IntSupplier shareGroupTopicPartitionCount);
+
+ /**
+ * Stop the share coordinator
+ */
+ void shutdown();
+
+ /**
+ * Handle write share state call
+ * @param context - represents the incoming write request context
Review Comment:
Do we require `-`, I see it's being used sometimes for params and omitted in some case. Shall we have consistency?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1726384651
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
+
+ /**
+ * Return the partition index for the given key.
+ *
+ * @param key key - groupId:topicId:partitionId.
+ * @return The partition index.
+ */
+ int partitionFor(String key);
+
+ /**
+ * Return the configuration properties of the share-group state topic.
+ *
+ * @return Properties of the share-group state topic.
+ */
+ Properties shareGroupStateTopicConfigs();
+
+ /**
+ * Start the share coordinator
+ *
+ * @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
+ */
+ void startup(IntSupplier shareGroupTopicPartitionCount);
+
+ /**
+ * Stop the share coordinator
+ */
+ void shutdown();
+
+ /**
+ * Handle write share state call
+ * @param context - represents the incoming write request context
Review Comment:
Following. All occurrences in PR
```
* @param key key - groupId:topicId:partitionId.
* @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
* @param context - represents the incoming write request context
* @param request - actual RPC request object
* @param context - represents the incoming read request context
* @param request - actual RPC request object
* @param partitionIndex - The partition index (internal topic)
* @param partitionLeaderEpoch - Leader epoch of the partition (internal topic)
* @param partitionIndex - The partition index (internal topic)
* @param partitionLeaderEpoch - Leader epoch of the partition (internal topic). Empty optional means deleted.
* @param recordType - short representing version
* @param recordType - short representing version
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #16921:
URL: https://github.com/apache/kafka/pull/16921#issuecomment-2304475463
@dajac @mumrah
Could you please review this PR as it is related to the coordinator?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #16921:
URL: https://github.com/apache/kafka/pull/16921#issuecomment-2304845881
unrelated test failures
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1729403619
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterOffsetsStateBatch.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.Objects;
+
+/**
+ * This is a helper class which overrides the equals and hashcode
+ * methods to only focus on the first and last offset fields of the
+ * state batch. This is useful when combining batches.
+ */
+public class PersisterOffsetsStateBatch {
+ private final PersisterStateBatch delegate;
+
+ public PersisterOffsetsStateBatch(long firstOffset, long lastOffset, byte deliveryState, short deliveryCount) {
Review Comment:
nit: break up this long constructor
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/PersisterOffsetsStateBatch.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.server.group.share.PersisterStateBatch;
+
+import java.util.Objects;
+
+/**
+ * This is a helper class which overrides the equals and hashcode
+ * methods to only focus on the first and last offset fields of the
+ * state batch. This is useful when combining batches.
+ */
+public class PersisterOffsetsStateBatch {
+ private final PersisterStateBatch delegate;
+
+ public PersisterOffsetsStateBatch(long firstOffset, long lastOffset, byte deliveryState, short deliveryCount) {
+ delegate = new PersisterStateBatch(firstOffset, lastOffset, deliveryState, deliveryCount);
+ }
+
+ public long firstOffset() {
+ return delegate.firstOffset();
+ }
+
+ public long lastOffset() {
+ return delegate.lastOffset();
+ }
+
+ public byte deliveryState() {
+ return delegate.deliveryState();
+ }
+
+ public short deliveryCount() {
+ return delegate.deliveryCount();
+ }
+
+ public static PersisterOffsetsStateBatch from(WriteShareGroupStateRequestData.StateBatch batch) {
+ return new PersisterOffsetsStateBatch(batch.firstOffset(), batch.lastOffset(), batch.deliveryState(), batch.deliveryCount());
Review Comment:
nit: break this long line
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1726383112
##########
share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorRecordSerdeTest.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorLoader;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotKey;
+import org.apache.kafka.coordinator.share.generated.ShareSnapshotValue;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateKey;
+import org.apache.kafka.coordinator.share.generated.ShareUpdateValue;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ShareCoordinatorRecordSerdeTest {
+ private ShareCoordinatorRecordSerde serde;
+
+ @BeforeEach
+ public void setup() {
+ serde = new ShareCoordinatorRecordSerde();
+ }
+
+ @Test
+ public void testSerializeKey() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+
+ assertArrayEquals(
+ MessageUtil.toVersionPrefixedBytes(record.key().version(), record.key().message()),
+ serde.serializeKey(record)
+ );
+ }
+
+ @Test
+ public void testSerializeValue() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+
+ assertArrayEquals(
+ MessageUtil.toVersionPrefixedBytes(record.value().version(), record.value().message()),
+ serde.serializeValue(record)
+ );
+ }
+
+ @Test
+ public void testSerializeNullValue() {
+ CoordinatorRecord record = new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("group")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ ),
+ null
+ );
+
+ assertNull(serde.serializeValue(record));
+ }
+
+ @Test
+ public void testDeserialize() {
+ CoordinatorRecord record = getShareSnapshotRecord("groupId", Uuid.randomUuid(), 1);
+ ApiMessageAndVersion key = record.key();
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ApiMessageAndVersion value = record.value();
+ ByteBuffer valueBuffer = MessageUtil.toVersionPrefixedByteBuffer(value.version(), value.message());
+
+ CoordinatorRecord desRecord = serde.deserialize(keyBuffer, valueBuffer);
+ assertEquals(key, desRecord.key());
+ assertEquals(value, desRecord.value());
+ }
+
+ @Test
+ public void testDeserializeWithTombstoneForValue() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("groupId")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ CoordinatorRecord record = serde.deserialize(keyBuffer, null);
+ assertEquals(key, record.key());
+ assertNull(record.value());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidRecordType() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(64);
+ keyBuffer.putShort((short) 255);
+ keyBuffer.rewind();
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(64);
+
+ CoordinatorLoader.UnknownRecordTypeException ex =
+ assertThrows(CoordinatorLoader.UnknownRecordTypeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals((short) 255, ex.unknownType());
+ }
+
+ @Test
+ public void testDeserializeWithKeyEmptyBuffer() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(0);
+ ByteBuffer valueBuffer = ByteBuffer.allocate(64);
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals("Could not read version from key's buffer.", ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithValueEmptyBuffer() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("foo")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(0);
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertEquals("Could not read version from value's buffer.", ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidKeyBytes() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(2);
+ keyBuffer.putShort((short) 0);
+ keyBuffer.rewind();
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(2);
+ valueBuffer.putShort((short) 0);
+ valueBuffer.rewind();
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertTrue(ex.getMessage().startsWith("Could not read record with version 0 from key's buffer due to"),
+ ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeWithInvalidValueBytes() {
+ ApiMessageAndVersion key = new ApiMessageAndVersion(
+ new ShareSnapshotKey()
+ .setGroupId("foo")
+ .setTopicId(Uuid.randomUuid())
+ .setPartition(1),
+ ShareCoordinator.SHARE_SNAPSHOT_RECORD_KEY_VERSION
+ );
+ ByteBuffer keyBuffer = MessageUtil.toVersionPrefixedByteBuffer(key.version(), key.message());
+
+ ByteBuffer valueBuffer = ByteBuffer.allocate(2);
+ valueBuffer.putShort((short) 0);
+ valueBuffer.rewind();
+
+ RuntimeException ex =
+ assertThrows(RuntimeException.class,
+ () -> serde.deserialize(keyBuffer, valueBuffer));
+ assertTrue(ex.getMessage().startsWith("Could not read record with version 0 from value's buffer due to"),
+ ex.getMessage());
+ }
+
+ @Test
+ public void testDeserializeAllRecordTypes() {
+ roundTrip((short) 0, new ShareSnapshotKey(), new ShareSnapshotValue());
Review Comment:
No it does not check that - we can pass any version and any key value pair. It will work. Same for group coord counderpart.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730849021
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
Review Comment:
But that will be needed for version changes from upstream.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730868431
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
Review Comment:
> But that will be needed for version changes from upstream.
yes, the scenario I imagined is there are some "new" messages produced by the new brokers. And then we run the older brokers to load those "new" messages. I assumed it should throw error likes `Unknown xxx Unknown`. WDYT? is it valid? in the old coordinator, it checks version range when loading group and offsets (https://github.com/apache/kafka/blob/4ae0ab38dd3a82cf2c36032014ee74af63251859/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala#L1183)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730578171
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
Review Comment:
Should we change the version defined in json (https://github.com/apache/kafka/blob/trunk/share-coordinator/src/main/resources/common/message/ShareUpdateValue.json#L21) to `1` for consistency?
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
+ } catch (RuntimeException ex) {
+ throw new RuntimeException(String.format("Could not read record with version %d from %s's buffer due to: %s.",
+ version, name, ex.getMessage()), ex);
+ }
+ }
+
+ /**
+ * Concrete child class must provide implementation which returns appropriate
+ * type of {@link ApiMessage} objects representing the key.
+ *
+ * @param recordType - short representing version
+ * @return ApiMessage object
+ */
+ protected abstract ApiMessage apiMessageKeyFor(short recordType);
Review Comment:
How about renaming `recordType` to `recordVersion`?
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
Review Comment:
I know the current value version is equal to key version, but we should pass `valueVersion` to avoid possible bugs in the future, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah merged PR #16921:
URL: https://github.com/apache/kafka/pull/16921
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1723973298
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
+
+ /**
+ * Return the partition index for the given key.
+ *
+ * @param key key - groupId:topicId:partitionId.
+ * @return The partition index.
+ */
+ int partitionFor(String key);
+
+ /**
+ * Return the configuration properties of the share-group state topic.
+ *
+ * @return Properties of the share-group state topic.
+ */
+ Properties shareGroupStateTopicConfigs();
+
+ /**
+ * Start the share coordinator
+ *
+ * @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
+ */
+ void startup(IntSupplier shareGroupTopicPartitionCount);
+
+ /**
+ * Stop the share coordinator
+ */
+ void shutdown();
+
+ /**
+ * Handle write share state call
+ * @param context - represents the incoming write request context
Review Comment:
code exists with and with `-` in the kafka codebase. The - has been used consistently in the interface in question. Refactoring entire kafka codebase to be consistent is beyond the scope of this PR.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.share;
+
+import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
+import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
+import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
+import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.requests.RequestContext;
+
+import java.util.OptionalInt;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntSupplier;
+
+public interface ShareCoordinator {
+ short SHARE_SNAPSHOT_RECORD_KEY_VERSION = 0;
+ short SHARE_SNAPSHOT_RECORD_VALUE_VERSION = 0;
+ short SHARE_UPDATE_RECORD_KEY_VERSION = 1;
+ short SHARE_UPDATE_RECORD_VALUE_VERSION = 1;
+
+ /**
+ * Return the partition index for the given key.
+ *
+ * @param key key - groupId:topicId:partitionId.
+ * @return The partition index.
+ */
+ int partitionFor(String key);
+
+ /**
+ * Return the configuration properties of the share-group state topic.
+ *
+ * @return Properties of the share-group state topic.
+ */
+ Properties shareGroupStateTopicConfigs();
+
+ /**
+ * Start the share coordinator
+ *
+ * @param shareGroupTopicPartitionCount - supplier returning the number of partitions for __share_group_state topic
+ */
+ void startup(IntSupplier shareGroupTopicPartitionCount);
+
+ /**
+ * Stop the share coordinator
+ */
+ void shutdown();
+
+ /**
+ * Handle write share state call
+ * @param context - represents the incoming write request context
Review Comment:
code exists with and without `-` in the kafka codebase. The - has been used consistently in the interface in question. Refactoring entire kafka codebase to be consistent is beyond the scope of this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17367: Share coordinator impl. Introduce infra classes [1/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #16921:
URL: https://github.com/apache/kafka/pull/16921#discussion_r1730849021
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRecordSerde.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.common.runtime;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
+ * <pre>
+ * record_key = [record_type key_message]
+ * record_value = [value_version value_message]
+ *
+ * record_type : The record type is currently define as the version of the key
+ * {@link ApiMessageAndVersion} object.
+ * key_message : The serialized message of the key {@link ApiMessageAndVersion} object.
+ * value_version : The value version is currently define as the version of the value
+ * {@link ApiMessageAndVersion} object.
+ * value_message : The serialized message of the value {@link ApiMessageAndVersion} object.
+ * </pre>
+ */
+public abstract class CoordinatorRecordSerde implements Serializer<CoordinatorRecord>, Deserializer<CoordinatorRecord> {
+ @Override
+ public byte[] serializeKey(CoordinatorRecord record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
+ record.key().version(),
+ record.key().message()
+ );
+ }
+
+ @Override
+ public byte[] serializeValue(CoordinatorRecord record) {
+ // Tombstone is represented with a null value.
+ if (record.value() == null) {
+ return null;
+ } else {
+ return MessageUtil.toVersionPrefixedBytes(
+ record.value().version(),
+ record.value().message()
+ );
+ }
+ }
+
+ @Override
+ public CoordinatorRecord deserialize(
+ ByteBuffer keyBuffer,
+ ByteBuffer valueBuffer
+ ) throws RuntimeException {
+ final short recordType = readVersion(keyBuffer, "key");
+ final ApiMessage keyMessage = apiMessageKeyFor(recordType);
+ readMessage(keyMessage, keyBuffer, recordType, "key");
+
+ if (valueBuffer == null) {
+ return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
+ }
+
+ final ApiMessage valueMessage = apiMessageValueFor(recordType);
+ final short valueVersion = readVersion(valueBuffer, "value");
+ readMessage(valueMessage, valueBuffer, valueVersion, "value");
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(keyMessage, recordType),
+ new ApiMessageAndVersion(valueMessage, valueVersion)
+ );
+ }
+
+ private short readVersion(ByteBuffer buffer, String name) throws RuntimeException {
+ try {
+ return buffer.getShort();
+ } catch (BufferUnderflowException ex) {
+ throw new RuntimeException(String.format("Could not read version from %s's buffer.", name));
+ }
+ }
+
+ private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
+ try {
+ message.read(new ByteBufferAccessor(buffer), version);
Review Comment:
The version checks are for the data fields and the "validVersions" record field.
Additionally, there are 2 methods hightestSupportedVersion and lowestSupportedVersion which specify the range
If the version is such that it is not handled by `apiMessageKeyFor` or `apiMessageValueFor`, it will throw the exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org