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; + [email protected] 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
