diff --git a/src/main/java/io/streamnative/kop/coordinator/group/GroupMetadata.java b/src/main/java/io/streamnative/kop/coordinator/group/GroupMetadata.java index 3108f859cd..14e848b062 100644 --- a/src/main/java/io/streamnative/kop/coordinator/group/GroupMetadata.java +++ b/src/main/java/io/streamnative/kop/coordinator/group/GroupMetadata.java @@ -23,6 +23,7 @@ import com.google.common.base.Supplier; import com.google.common.collect.Sets; import io.streamnative.kop.coordinator.group.MemberMetadata.MemberSummary; +import io.streamnative.kop.offset.OffsetAndMetadata; import io.streamnative.kop.utils.CoreUtils; import java.util.Collections; import java.util.Comparator; @@ -37,12 +38,16 @@ import java.util.UUID; import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; +import java.util.stream.Stream; import javax.annotation.concurrent.NotThreadSafe; import lombok.Data; import lombok.Getter; import lombok.Setter; import lombok.experimental.Accessors; +import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; +import org.apache.kafka.common.TopicPartition; +import org.apache.pulsar.common.schema.KeyValue; /** * Group contains the following metadata: @@ -60,6 +65,7 @@ @NotThreadSafe @Setter @Accessors(fluent = true) +@Slf4j class GroupMetadata { private static final Map> validPreviousStates = new HashMap<>(); @@ -147,6 +153,22 @@ static class GroupSummary { private final List members; } + /** + * We cache offset commits along with their commit record offset. This enables us to ensure that the latest offset + * commit is always materialized when we have a mix of transactional and regular offset commits. Without preserving + * information of the commit record offset, compaction of the offsets topic it self may result in the wrong offset + * commit being materialized. + */ + @Data + static class CommitRecordMetadataAndOffset { + private final Optional appendedBatchOffset; + private final OffsetAndMetadata offsetAndMetadata; + + public boolean olderThan(CommitRecordMetadataAndOffset that) { + return appendedBatchOffset.get() < that.appendedBatchOffset.get(); + } + } + private final String groupId; @Getter private final ReentrantLock lock = new ReentrantLock(); @@ -161,6 +183,12 @@ static class GroupSummary { // state management private final Map members = new HashMap<>(); + private final Map offsets = new HashMap<>(); + private final Map pendingOffsetCommits = new HashMap<>(); + private final Map> pendingTransactionalOffsetCommits = + new HashMap<>(); + private boolean receivedTransactionalOffsetCommits = false; + private boolean receivedConsumerOffsetCommits = false; GroupMetadata(String groupId, GroupState initialState) { this.groupId = groupId; @@ -389,6 +417,223 @@ public GroupOverview overview() { ); } + public void initializeOffsets(Map offsets, + Map> pendingTxnOffsets) { + this.offsets.putAll(offsets); + this.pendingTransactionalOffsetCommits.putAll(pendingTxnOffsets); + } + + public void onOffsetCommitAppend(TopicPartition topicPartition, + CommitRecordMetadataAndOffset offsetWithCommitRecordMetadata) { + if (pendingOffsetCommits.containsKey(topicPartition)) { + if (!offsetWithCommitRecordMetadata.appendedBatchOffset.isPresent()) { + throw new IllegalStateException("Cannot complete offset commit write without providing the metadata" + + " of the record in the log."); + } + if (!offsets.containsKey(topicPartition) + || offsets.get(topicPartition).olderThan(offsetWithCommitRecordMetadata)) { + offsets.put(topicPartition, offsetWithCommitRecordMetadata); + } + } + + OffsetAndMetadata stagedOffset = pendingOffsetCommits.get(topicPartition); + if (null != stagedOffset && offsetWithCommitRecordMetadata.offsetAndMetadata == stagedOffset) { + pendingOffsetCommits.remove(topicPartition); + } else { + // The pendingOffsetCommits for this partition could be empty if the topic was deleted, in which case + // its entries would be removed from the cache by the `removeOffsets` method. + } + } + + public void failPendingOffsetWrite(TopicPartition topicPartition, + OffsetAndMetadata offset) { + OffsetAndMetadata pendingOffset = pendingOffsetCommits.get(topicPartition); + if (pendingOffset != null && offset == pendingOffset) { + pendingOffsetCommits.remove(topicPartition); + } + } + + public void prepareOffsetCommit(Map offsets) { + receivedConsumerOffsetCommits = true; + pendingOffsetCommits.putAll(offsets); + } + + public void prepareTxnOffsetCommit(long producerId, + Map offsets) { + if (log.isTraceEnabled()) { + log.trace("TxnOffsetCommit for producer {} and group {} with offsets {} is pending", + producerId, groupId, offsets); + } + receivedTransactionalOffsetCommits = true; + Map producerOffsets = + pendingTransactionalOffsetCommits.computeIfAbsent(producerId, pid -> new HashMap<>()); + offsets.forEach((tp, offsetsAndMetadata) -> producerOffsets.put(tp, new CommitRecordMetadataAndOffset( + Optional.empty(), + offsetsAndMetadata + ))); + } + + public boolean hasReceivedConsistentOffsetCommits() { + return !receivedConsumerOffsetCommits || !receivedTransactionalOffsetCommits; + } + + /** + * Remove a pending transactional offset commit if the actual offset commit record was not written to the log. + * We will return an error and the client will retry the request, potentially to a different coordinator. + */ + public void failPendingTxnOffsetCommit(long producerId, + TopicPartition topicPartition) { + Map pendingOffsets = + pendingTransactionalOffsetCommits.get(producerId); + if (null != pendingOffsets) { + CommitRecordMetadataAndOffset pendingOffsetCommit = pendingOffsets.remove(topicPartition); + if (log.isTraceEnabled()) { + log.trace("TxnOffsetCommit for producer {} and group {} with offsets {} failed to be appended" + + " to the log", + producerId, groupId, pendingOffsetCommit); + } + if (pendingOffsets.isEmpty()) { + pendingTransactionalOffsetCommits.remove(producerId); + } + } else { + // We may hit this case if the partition in question has emigrated already. + } + } + + public void onTxnOffsetCommitAppend(long producerId, + TopicPartition topicPartition, + CommitRecordMetadataAndOffset commitRecordMetadataAndOffset) { + Map pendingOffsets = + pendingTransactionalOffsetCommits.get(producerId); + if (null != pendingOffsets) { + if (pendingOffsets.containsKey(topicPartition) + && pendingOffsets.get(topicPartition).offsetAndMetadata() + == commitRecordMetadataAndOffset.offsetAndMetadata) { + pendingOffsets.put(topicPartition, commitRecordMetadataAndOffset); + } + } else { + // We may hit this case if the partition in question has emigrated. + } + } + + /** + * Complete a pending transactional offset commit. This is called after a commit or abort marker is fully written + * to the log. + */ + public void completePendingTxnOffsetCommit(long producerId, + boolean isCommit) { + Map pendingOffsets = + pendingTransactionalOffsetCommits.remove(producerId); + if (isCommit) { + if (null != pendingOffsets) { + pendingOffsets.entrySet().forEach(e -> { + TopicPartition topicPartition = e.getKey(); + CommitRecordMetadataAndOffset commitRecordMetadataAndOffset = e.getValue(); + if (!commitRecordMetadataAndOffset.appendedBatchOffset.isPresent()) { + throw new IllegalStateException(String.format("Trying to complete a transactional offset" + + " commit for producerId %s and groupId %s even though the offset commit record" + + " itself hasn't been appended to the log.", producerId, groupId)); + } + + CommitRecordMetadataAndOffset currentOffsetOpt = offsets.get(topicPartition); + if (currentOffsetOpt == null || currentOffsetOpt.olderThan(commitRecordMetadataAndOffset)) { + if (log.isTraceEnabled()) { + log.trace("TxnOffsetCommit for producer {} and group {} with offset {} " + + "committed and loaded into the cache.", + producerId, groupId, commitRecordMetadataAndOffset); + } + offsets.put(topicPartition, commitRecordMetadataAndOffset); + } else { + if (log.isTraceEnabled()) { + log.trace("TxnOffsetCommit for producer {} and group {} with offset {} " + + "committed, but not loaded since its offset is older than current offset" + + " {}.", + producerId, groupId, commitRecordMetadataAndOffset, currentOffsetOpt); + } + } + }); + } + } else { + if (log.isTraceEnabled()) { + log.trace("TxnOffsetCommit for producer {} and group {} with offsets {} aborted", + producerId, groupId, pendingOffsets); + } + } + } + + public Set activeProducers() { + return pendingTransactionalOffsetCommits.keySet(); + } + + public boolean hasPendingOffsetCommitsFromProducer(long producerId) { + return pendingTransactionalOffsetCommits.containsKey(producerId); + } + + public Map removeAllOffsets() { + return removeOffsets(offsets.keySet().stream()); + } + + public Map removeOffsets(Stream topicPartitions) { + return topicPartitions.map(topicPartition -> { + pendingOffsetCommits.remove(topicPartition); + pendingTransactionalOffsetCommits.forEach((pid, pendingOffsets) -> { + pendingOffsets.remove(topicPartition); + }); + CommitRecordMetadataAndOffset removedOffset = offsets.remove(topicPartition); + return new KeyValue<>( + topicPartition, + removedOffset.offsetAndMetadata() + ); + }).collect(Collectors.toMap( + e -> e.getKey(), + e -> e.getValue() + )); + } + + public Map removeExpiredOffsets(long startMs) { + Map expiredOffsets = offsets.entrySet().stream() + .filter(e -> + e.getValue().offsetAndMetadata().expireTimestamp() < startMs + && !pendingOffsetCommits.containsKey(e.getKey())) + .map(e -> new KeyValue<>( + e.getKey(), + e.getValue().offsetAndMetadata() + )) + .collect(Collectors.toMap( + kv -> kv.getKey(), + kv -> kv.getValue() + )); + + expiredOffsets.keySet().forEach(tp -> offsets.remove(tp)); + return expiredOffsets; + } + + public Map allOffsets() { + return offsets.entrySet().stream().collect(Collectors.toMap( + e -> e.getKey(), + e -> e.getValue().offsetAndMetadata() + )); + } + + public Optional offset(TopicPartition topicPartition) { + return Optional.ofNullable(offsets.get(topicPartition)).map(e -> e.offsetAndMetadata); + } + + // visible for testing + Optional offsetWithRecordMetadata(TopicPartition topicPartition) { + return Optional.ofNullable(offsets.get(topicPartition)); + } + + public int numOffsets() { + return offsets.size(); + } + + public boolean hasOffsets() { + return !offsets.isEmpty() + || !pendingOffsetCommits.isEmpty() + || !pendingTransactionalOffsetCommits.isEmpty(); + } + @Override public String toString() { ToStringHelper helper = MoreObjects.toStringHelper("GroupMetadata") diff --git a/src/main/java/io/streamnative/kop/coordinator/group/GroupMetadataManager.java b/src/main/java/io/streamnative/kop/coordinator/group/GroupMetadataManager.java index ba151240ab..2eedbda527 100644 --- a/src/main/java/io/streamnative/kop/coordinator/group/GroupMetadataManager.java +++ b/src/main/java/io/streamnative/kop/coordinator/group/GroupMetadataManager.java @@ -22,7 +22,9 @@ import static java.nio.charset.StandardCharsets.UTF_8; import static org.apache.kafka.common.internals.Topic.GROUP_METADATA_TOPIC_NAME; +import io.streamnative.kop.offset.OffsetAndMetadata; import java.nio.ByteBuffer; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.Map; @@ -41,6 +43,7 @@ import org.apache.bookkeeper.common.util.MathUtils; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.record.RecordBatch; import org.apache.kafka.common.utils.Time; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; @@ -104,6 +107,7 @@ public String toString() { } + private final OffsetConfig config; private final ConcurrentMap groupMetadataCache; /* lock protecting access to loading and owned partition sets */ private final ReentrantLock partitionLock = new ReentrantLock(); @@ -117,14 +121,25 @@ public String toString() { /* shutting down flag */ private final AtomicBoolean shuttingDown = new AtomicBoolean(false); private final int groupMetadataTopicPartitionCount; + + /** + * The groups with open transactional offsets commits per producer. We need this because when the commit or abort + * marker comes in for a transaction, it is for a particular partition on the offsets topic and a particular + * producerId. We use this structure to quickly find the groups which need to be updated by the commit/abort + * marker. + */ + private final Map> openGroupsForProducer = new HashMap<>(); + private final Producer metadataTopicProducer; private final Reader metadataTopicReader; private final Time time; GroupMetadataManager(int groupMetadataTopicPartitionCount, + OffsetConfig config, Producer metadataTopicProducer, Reader metadataTopicConsumer, Time time) { + this.config = config; this.groupMetadataCache = new ConcurrentHashMap<>(); this.groupMetadataTopicPartitionCount = groupMetadataTopicPartitionCount; this.metadataTopicProducer = metadataTopicProducer; @@ -183,6 +198,13 @@ && getGroup(groupId) ); } + boolean isGroupOpenForProducer(long producerId, + String groupId) { + return openGroupsForProducer.getOrDefault( + producerId, Collections.emptySet() + ).contains(groupId); + } + public Optional getGroup(String groupId) { return Optional.ofNullable(groupMetadataCache.getOrDefault(groupId, null)); } @@ -212,6 +234,46 @@ public CompletableFuture storeGroup(GroupMetadata group, .exceptionally(cause -> Errors.COORDINATOR_NOT_AVAILABLE); } + public CompletableFuture> storeOffsets( + GroupMetadata group, + String consumerId, + Map offsetMetadata + ) { + return storeOffsets( + group, + consumerId, + offsetMetadata, + RecordBatch.NO_PRODUCER_ID, + RecordBatch.NO_PRODUCER_EPOCH + ); + } + + public CompletableFuture> storeOffsets( + GroupMetadata group, + String consumerId, + Map offsetMetadata, + long producerId, + short producerEpoch + ) { + // first filter out partitions with offset metadata size exceeding limit + // Map filteredOffsetMetadata = + // offsetMetadata.entrySet().stream() + // .filter(entry -> validateOffsetMetadataLength(entry.getValue().metadata())) + // .collect(Collectors.toMap( + // e -> e.getKey(), + // e -> e.getValue() + // )); + + throw new UnsupportedOperationException(); + } + + /* + * Check if the offset metadata length is valid + */ + private boolean validateOffsetMetadataLength(String metadata) { + return metadata == null || metadata.length() <= config.maxMetadataSize(); + } + public CompletableFuture scheduleLoadGroupAndOffsets(int offsetsPartition, Consumer onGroupLoaded) { TopicPartition topicPartition = new TopicPartition( diff --git a/src/main/java/io/streamnative/kop/coordinator/group/OffsetConfig.java b/src/main/java/io/streamnative/kop/coordinator/group/OffsetConfig.java index cbf172480f..aaf4ba4c94 100644 --- a/src/main/java/io/streamnative/kop/coordinator/group/OffsetConfig.java +++ b/src/main/java/io/streamnative/kop/coordinator/group/OffsetConfig.java @@ -14,6 +14,7 @@ package io.streamnative.kop.coordinator.group; import lombok.Builder; +import lombok.Builder.Default; import lombok.Data; import lombok.experimental.Accessors; import org.apache.kafka.common.record.CompressionType; @@ -26,6 +27,11 @@ @Accessors(fluent = true) public class OffsetConfig { - private final CompressionType offsetsTopicCompressionType; + private static final int DefaultMaxMetadataSize = 4096; + + @Default + private int maxMetadataSize = DefaultMaxMetadataSize; + @Default + private CompressionType offsetsTopicCompressionType = CompressionType.NONE; } diff --git a/src/main/java/io/streamnative/kop/offset/OffsetAndMetadata.java b/src/main/java/io/streamnative/kop/offset/OffsetAndMetadata.java new file mode 100644 index 0000000000..da7a5857f5 --- /dev/null +++ b/src/main/java/io/streamnative/kop/offset/OffsetAndMetadata.java @@ -0,0 +1,103 @@ +/** + * Licensed 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 io.streamnative.kop.offset; + +import static org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP; + +import lombok.AccessLevel; +import lombok.Data; +import lombok.RequiredArgsConstructor; +import lombok.experimental.Accessors; + +/** + * Offset and metadata. + */ +@Data +@Accessors(fluent = true) +@RequiredArgsConstructor(access = AccessLevel.PRIVATE) +public class OffsetAndMetadata { + + public static OffsetAndMetadata apply( + long offset, + String metadata, + long commitTimestamp, + long expireTimestamp + ) { + return new OffsetAndMetadata( + new OffsetMetadata(offset, metadata), + commitTimestamp, + expireTimestamp + ); + } + + public static OffsetAndMetadata apply( + long offset, + String metadata, + long timestamp + ) { + return new OffsetAndMetadata( + new OffsetMetadata(offset, metadata), + timestamp, + timestamp + ); + } + + public static OffsetAndMetadata apply( + long offset, + String metadata + ) { + return new OffsetAndMetadata( + new OffsetMetadata(offset, metadata) + ); + } + + public static OffsetAndMetadata apply( + long offset + ) { + return new OffsetAndMetadata( + new OffsetMetadata(offset, OffsetMetadata.NO_METADATA) + ); + } + + private final OffsetMetadata offsetMetadata; + private final long commitTimestamp; + private final long expireTimestamp; + + @SuppressWarnings("deprecation") + private OffsetAndMetadata(OffsetMetadata offsetMetadata) { + this( + offsetMetadata, + DEFAULT_TIMESTAMP, + DEFAULT_TIMESTAMP); + } + + public long offset() { + return offsetMetadata.offset(); + } + + public String metadata() { + return offsetMetadata.metadata(); + } + + @Override + public String toString() { + return String.format( + "[%s,CommitTime %d,ExpirationTime %d]", + offsetMetadata, + commitTimestamp, + expireTimestamp + ); + } + +} diff --git a/src/main/java/io/streamnative/kop/offset/OffsetMetadata.java b/src/main/java/io/streamnative/kop/offset/OffsetMetadata.java new file mode 100644 index 0000000000..d00099ed72 --- /dev/null +++ b/src/main/java/io/streamnative/kop/offset/OffsetMetadata.java @@ -0,0 +1,52 @@ +/** + * Licensed 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 io.streamnative.kop.offset; + +import lombok.Data; +import lombok.experimental.Accessors; + +/** + * Offset Metadata. + */ +@Data +@Accessors(fluent = true) +public class OffsetMetadata { + + public static final long INVALID_OFFSET = -1L; + public static final String NO_METADATA = ""; + public static final OffsetMetadata INVALID_OFFSET_METADATA = + new OffsetMetadata(INVALID_OFFSET, NO_METADATA); + + private final long offset; + private final String metadata; + + public OffsetMetadata(long offset) { + this(offset, NO_METADATA); + } + + public OffsetMetadata(long offset, String metadata) { + this.offset = offset; + this.metadata = metadata; + } + + @Override + public String toString() { + return String.format( + "OffsetMetadata[%d,%s]", + offset, + metadata != null && metadata.length() > 0 ? metadata : "NO_METADATA" + ); + } + +} diff --git a/src/main/java/io/streamnative/kop/offset/OffsetMetadataAndError.java b/src/main/java/io/streamnative/kop/offset/OffsetMetadataAndError.java new file mode 100644 index 0000000000..44dcdec5dd --- /dev/null +++ b/src/main/java/io/streamnative/kop/offset/OffsetMetadataAndError.java @@ -0,0 +1,89 @@ +/** + * Licensed 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 io.streamnative.kop.offset; + +import lombok.AccessLevel; +import lombok.Data; +import lombok.RequiredArgsConstructor; +import org.apache.kafka.common.protocol.Errors; + +/** + * Offset metadata and errors. + */ +@Data +@RequiredArgsConstructor(access = AccessLevel.PRIVATE) +public class OffsetMetadataAndError { + + public static final OffsetMetadataAndError NO_OFFSET = + new OffsetMetadataAndError(OffsetMetadata.INVALID_OFFSET_METADATA, Errors.NONE); + public static final OffsetMetadataAndError GROUP_LOADING = + new OffsetMetadataAndError(OffsetMetadata.INVALID_OFFSET_METADATA, Errors.COORDINATOR_LOAD_IN_PROGRESS); + public static final OffsetMetadataAndError UNKNOWN_MEMBER = + new OffsetMetadataAndError(OffsetMetadata.INVALID_OFFSET_METADATA, Errors.UNKNOWN_MEMBER_ID); + public static final OffsetMetadataAndError NOT_COORDINATOR_FOR_GROUP = + new OffsetMetadataAndError(OffsetMetadata.INVALID_OFFSET_METADATA, Errors.NOT_COORDINATOR); + public static final OffsetMetadataAndError GROUP_COORDINATOR_NOT_AVAILABLE = + new OffsetMetadataAndError(OffsetMetadata.INVALID_OFFSET_METADATA, Errors.COORDINATOR_NOT_AVAILABLE); + public static final OffsetMetadataAndError UNKNOWN_TOPIC_OR_PARTITION = + new OffsetMetadataAndError(OffsetMetadata.INVALID_OFFSET_METADATA, Errors.UNKNOWN_TOPIC_OR_PARTITION); + public static final OffsetMetadataAndError ILLEGAL_GROUP_GENERATION_ID = + new OffsetMetadataAndError(OffsetMetadata.INVALID_OFFSET_METADATA, Errors.ILLEGAL_GENERATION); + + public static OffsetMetadataAndError apply(long offset) { + return new OffsetMetadataAndError( + new OffsetMetadata(offset, OffsetMetadata.NO_METADATA), + Errors.NONE + ); + } + + public static OffsetMetadataAndError apply(Errors errors) { + return new OffsetMetadataAndError( + OffsetMetadata.INVALID_OFFSET_METADATA, + errors + ); + } + + public static OffsetMetadataAndError apply(long offset, + String metadata, + Errors errors) { + return new OffsetMetadataAndError( + new OffsetMetadata(offset, metadata), + errors + ); + } + + private final OffsetMetadata offsetMetadata; + private final Errors error; + + private OffsetMetadataAndError(OffsetMetadata offsetMetadata) { + this(offsetMetadata, Errors.NONE); + } + + public long offset() { + return offsetMetadata.offset(); + } + + public String metadata() { + return offsetMetadata.metadata(); + } + + @Override + public String toString() { + return String.format( + "[%s, Error=%s]", + offsetMetadata, + error + ); + } +} diff --git a/src/main/java/io/streamnative/kop/offset/package-info.java b/src/main/java/io/streamnative/kop/offset/package-info.java new file mode 100644 index 0000000000..b091d4390f --- /dev/null +++ b/src/main/java/io/streamnative/kop/offset/package-info.java @@ -0,0 +1,17 @@ +/** + * Licensed 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. + */ +/** + * Kafka Offset related classes. + */ +package io.streamnative.kop.offset; \ No newline at end of file diff --git a/src/test/java/io/streamnative/kop/coordinator/group/GroupMetadataManagerTest.java b/src/test/java/io/streamnative/kop/coordinator/group/GroupMetadataManagerTest.java index 3b2e4b3911..4880ed272b 100644 --- a/src/test/java/io/streamnative/kop/coordinator/group/GroupMetadataManagerTest.java +++ b/src/test/java/io/streamnative/kop/coordinator/group/GroupMetadataManagerTest.java @@ -64,6 +64,7 @@ public class GroupMetadataManagerTest extends MockKafkaServiceBaseTest { GroupMetadataManager groupMetadataManager = null; Producer producer = null; Reader consumer = null; + OffsetConfig offsetConfig = OffsetConfig.builder().build(); @Before @Override @@ -83,6 +84,7 @@ public void setup() throws Exception { time = new MockTime(); groupMetadataManager = new GroupMetadataManager( 1, + offsetConfig, producer, consumer, time @@ -154,6 +156,7 @@ void runGroupMetadataManagerTester(final String topicName, .create(); groupMetadataManager = new GroupMetadataManager( 1, + offsetConfig, producer, reader, time diff --git a/src/test/java/io/streamnative/kop/coordinator/group/GroupMetadataTest.java b/src/test/java/io/streamnative/kop/coordinator/group/GroupMetadataTest.java index adb727f3df..145e5de8f8 100644 --- a/src/test/java/io/streamnative/kop/coordinator/group/GroupMetadataTest.java +++ b/src/test/java/io/streamnative/kop/coordinator/group/GroupMetadataTest.java @@ -25,11 +25,16 @@ import static org.junit.Assert.fail; import com.google.common.collect.Sets; +import io.streamnative.kop.coordinator.group.GroupMetadata.CommitRecordMetadataAndOffset; +import io.streamnative.kop.offset.OffsetAndMetadata; +import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletableFuture; import lombok.val; +import org.apache.kafka.common.TopicPartition; import org.junit.Before; import org.junit.Test; @@ -377,6 +382,216 @@ public void testInitNextGenerationEmptyGroup() { assertNull(group.protocolOrNull()); } + @Test + public void testOffsetCommit() { + TopicPartition partition = new TopicPartition("foo", 0); + OffsetAndMetadata offset = OffsetAndMetadata.apply(37); + long commitRecordOffset = 3; + + Map offsets = new HashMap<>(); + offsets.put(partition, offset); + group.prepareOffsetCommit(offsets); + assertTrue(group.hasOffsets()); + assertEquals(Optional.empty(), group.offset(partition)); + + group.onOffsetCommitAppend( + partition, + new CommitRecordMetadataAndOffset(Optional.of(commitRecordOffset), offset)); + assertTrue(group.hasOffsets()); + assertEquals(Optional.of(offset), group.offset(partition)); + } + + @Test + public void testOffsetCommitFailure() { + TopicPartition partition = new TopicPartition("foo", 0); + OffsetAndMetadata offset = OffsetAndMetadata.apply(37); + + Map offsets = new HashMap<>(); + offsets.put(partition, offset); + group.prepareOffsetCommit(offsets); + assertTrue(group.hasOffsets()); + assertEquals(Optional.empty(), group.offset(partition)); + + group.failPendingOffsetWrite(partition, offset); + assertFalse(group.hasOffsets()); + assertEquals(Optional.empty(), group.offset(partition)); + } + + @Test + public void testOffsetCommitFailureWithAnotherPending() { + TopicPartition partition = new TopicPartition("foo", 0); + OffsetAndMetadata firstOffset = OffsetAndMetadata.apply(37); + OffsetAndMetadata secondOffset = OffsetAndMetadata.apply(57); + + Map offsets = new HashMap<>(); + offsets.put(partition, firstOffset); + group.prepareOffsetCommit(offsets); + assertTrue(group.hasOffsets()); + assertEquals(Optional.empty(), group.offset(partition)); + + offsets = new HashMap<>(); + offsets.put(partition, secondOffset); + group.prepareOffsetCommit(offsets); + assertTrue(group.hasOffsets()); + + group.failPendingOffsetWrite(partition, firstOffset); + assertTrue(group.hasOffsets()); + assertEquals(Optional.empty(), group.offset(partition)); + + group.onOffsetCommitAppend(partition, new CommitRecordMetadataAndOffset(Optional.of(3L), secondOffset)); + assertTrue(group.hasOffsets()); + assertEquals(Optional.of(secondOffset), group.offset(partition)); + } + + @Test + public void testOffsetCommitWithAnotherPending() { + TopicPartition partition = new TopicPartition("foo", 0); + OffsetAndMetadata firstOffset = OffsetAndMetadata.apply(37); + OffsetAndMetadata secondOffset = OffsetAndMetadata.apply(57); + + Map offsets = new HashMap<>(); + offsets.put(partition, firstOffset); + group.prepareOffsetCommit(offsets); + assertTrue(group.hasOffsets()); + assertEquals(Optional.empty(), group.offset(partition)); + + offsets = new HashMap<>(); + offsets.put(partition, secondOffset); + group.prepareOffsetCommit(offsets); + assertTrue(group.hasOffsets()); + + group.onOffsetCommitAppend(partition, new CommitRecordMetadataAndOffset(Optional.of(4L), firstOffset)); + assertTrue(group.hasOffsets()); + assertEquals(Optional.of(firstOffset), group.offset(partition)); + + group.onOffsetCommitAppend(partition, new CommitRecordMetadataAndOffset(Optional.of(5L), secondOffset)); + assertTrue(group.hasOffsets()); + assertEquals(Optional.of(secondOffset), group.offset(partition)); + } + + @Test + public void testConsumerBeatsTransactionalOffsetCommit() { + TopicPartition partition = new TopicPartition("foo", 0); + long producerId = 13232L; + OffsetAndMetadata txnOffsetCommit = OffsetAndMetadata.apply(37); + OffsetAndMetadata consumerOffsetCommit = OffsetAndMetadata.apply(57); + + Map offsets = new HashMap<>(); + offsets.put(partition, txnOffsetCommit); + group.prepareTxnOffsetCommit(producerId, offsets); + assertTrue(group.hasOffsets()); + assertEquals(Optional.empty(), group.offset(partition)); + + offsets = new HashMap<>(); + offsets.put(partition, consumerOffsetCommit); + group.prepareOffsetCommit(offsets); + assertTrue(group.hasOffsets()); + + group.onTxnOffsetCommitAppend(producerId, partition, + new CommitRecordMetadataAndOffset(Optional.of(3L), txnOffsetCommit)); + group.onOffsetCommitAppend(partition, + new CommitRecordMetadataAndOffset(Optional.of(4L), consumerOffsetCommit)); + assertTrue(group.hasOffsets()); + assertEquals(Optional.of(consumerOffsetCommit), group.offset(partition)); + + group.completePendingTxnOffsetCommit(producerId, true); + assertTrue(group.hasOffsets()); + // This is the crucial assertion which validates that we materialize offsets in offset order, + // not transactional order. + assertEquals(Optional.of(consumerOffsetCommit), group.offset(partition)); + } + + @Test + public void testTransactionBeatsConsumerOffsetCommit() { + TopicPartition partition = new TopicPartition("foo", 0); + long producerId = 13232L; + OffsetAndMetadata txnOffsetCommit = OffsetAndMetadata.apply(37); + OffsetAndMetadata consumerOffsetCommit = OffsetAndMetadata.apply(57); + + Map offsets = new HashMap<>(); + offsets.put(partition, txnOffsetCommit); + group.prepareTxnOffsetCommit(producerId, offsets); + assertTrue(group.hasOffsets()); + assertEquals(Optional.empty(), group.offset(partition)); + + offsets = new HashMap<>(); + offsets.put(partition, consumerOffsetCommit); + group.prepareOffsetCommit(offsets); + assertTrue(group.hasOffsets()); + + group.onOffsetCommitAppend( + partition, new CommitRecordMetadataAndOffset(Optional.of(3L), consumerOffsetCommit)); + group.onTxnOffsetCommitAppend(producerId, partition, + new CommitRecordMetadataAndOffset(Optional.of(4L), txnOffsetCommit)); + assertTrue(group.hasOffsets()); + // The transactional offset commit hasn't been committed yet, so we should materialize + // the consumer offset commit. + assertEquals(Optional.of(consumerOffsetCommit), group.offset(partition)); + + group.completePendingTxnOffsetCommit(producerId, true); + assertTrue(group.hasOffsets()); + // The transactional offset commit has been materialized and the transactional commit record + // is later in the log, so it should be materialized. + assertEquals(Optional.of(txnOffsetCommit), group.offset(partition)); + } + + @Test + public void testTransactionalCommitIsAbortedAndConsumerCommitWins() { + TopicPartition partition = new TopicPartition("foo", 0); + long producerId = 13232L; + OffsetAndMetadata txnOffsetCommit = OffsetAndMetadata.apply(37); + OffsetAndMetadata consumerOffsetCommit = OffsetAndMetadata.apply(57); + + Map offsets = new HashMap<>(); + offsets.put(partition, txnOffsetCommit); + group.prepareTxnOffsetCommit(producerId, offsets); + assertTrue(group.hasOffsets()); + assertEquals(Optional.empty(), group.offset(partition)); + + offsets = new HashMap<>(); + offsets.put(partition, consumerOffsetCommit); + group.prepareOffsetCommit(offsets); + assertTrue(group.hasOffsets()); + + group.onOffsetCommitAppend(partition, + new CommitRecordMetadataAndOffset(Optional.of(3L), consumerOffsetCommit)); + group.onTxnOffsetCommitAppend(producerId, partition, + new CommitRecordMetadataAndOffset(Optional.of(4L), txnOffsetCommit)); + assertTrue(group.hasOffsets()); + // The transactional offset commit hasn't been committed yet, so we should materialize the consumer + // offset commit. + assertEquals(Optional.of(consumerOffsetCommit), group.offset(partition)); + + group.completePendingTxnOffsetCommit(producerId, false); + assertTrue(group.hasOffsets()); + // The transactional offset commit should be discarded and the consumer offset commit should continue to be + // materialized. + assertFalse(group.hasPendingOffsetCommitsFromProducer(producerId)); + assertEquals(Optional.of(consumerOffsetCommit), group.offset(partition)); + } + + @Test + public void testFailedTxnOffsetCommitLeavesNoPendingState() { + TopicPartition partition = new TopicPartition("foo", 0); + long producerId = 13232L; + OffsetAndMetadata txnOffsetCommit = OffsetAndMetadata.apply(37); + + Map offsets = new HashMap<>(); + offsets.put(partition, txnOffsetCommit); + group.prepareTxnOffsetCommit(producerId, offsets); + assertTrue(group.hasPendingOffsetCommitsFromProducer(producerId)); + assertTrue(group.hasOffsets()); + assertEquals(Optional.empty(), group.offset(partition)); + group.failPendingTxnOffsetCommit(producerId, partition); + assertFalse(group.hasOffsets()); + assertFalse(group.hasPendingOffsetCommitsFromProducer(producerId)); + + // The commit marker should now have no effect. + group.completePendingTxnOffsetCommit(producerId, true); + assertFalse(group.hasOffsets()); + assertFalse(group.hasPendingOffsetCommitsFromProducer(producerId)); + } + private void assertState(GroupMetadata group, GroupState targetState) { Set states = Sets.newHashSet( Stable, PreparingRebalance, CompletingRebalance, Dead @@ -387,5 +602,4 @@ private void assertState(GroupMetadata group, GroupState targetState) { assertTrue(group.is(targetState)); } - }