Browse Source
Implement the QuorumController side of KRaft metadata transactions. As specified in KIP-868, this PR creates a new metadata version, IBP_3_6_IV1, which contains the three new records: AbortTransactionRecord, BeginTransactionRecord, EndTransactionRecord. In order to make offset management unit-testable, this PR moves it out of QuorumController.java and into OffsetControlManager.java. The general approach here is to track the "last stable offset," which is calculated by looking at the latest committed offset and the in-progress transaction (if any). When a transaction is aborted, we revert back to this last stable offset. We also revert back to it when the controller is transitioning from active to inactive. In a follow-up PR, we will add support for the transaction records in MetadataLoader. We will also add support for automatically aborting pending transactions after a controller failover. Reviewers: David Arthur <mumrah@gmail.com>pull/14209/head
Colin Patrick McCabe
1 year ago
committed by
GitHub
15 changed files with 913 additions and 159 deletions
@ -0,0 +1,421 @@
@@ -0,0 +1,421 @@
|
||||
/* |
||||
* 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.controller; |
||||
|
||||
import org.apache.kafka.common.metadata.AbortTransactionRecord; |
||||
import org.apache.kafka.common.metadata.BeginTransactionRecord; |
||||
import org.apache.kafka.common.metadata.EndTransactionRecord; |
||||
import org.apache.kafka.common.utils.LogContext; |
||||
import org.apache.kafka.common.utils.Time; |
||||
import org.apache.kafka.controller.metrics.QuorumControllerMetrics; |
||||
import org.apache.kafka.raft.Batch; |
||||
import org.apache.kafka.raft.OffsetAndEpoch; |
||||
import org.apache.kafka.server.common.ApiMessageAndVersion; |
||||
import org.apache.kafka.snapshot.Snapshots; |
||||
import org.apache.kafka.timeline.SnapshotRegistry; |
||||
import org.slf4j.Logger; |
||||
|
||||
import java.util.Optional; |
||||
|
||||
|
||||
/** |
||||
* Manages read and write offsets, and in-memory snapshots. |
||||
* |
||||
* Also manages the following metrics: |
||||
* kafka.controller:type=KafkaController,name=ActiveControllerCount |
||||
* kafka.controller:type=KafkaController,name=LastAppliedRecordLagMs |
||||
* kafka.controller:type=KafkaController,name=LastAppliedRecordOffset |
||||
* kafka.controller:type=KafkaController,name=LastAppliedRecordTimestamp |
||||
* kafka.controller:type=KafkaController,name=LastCommittedRecordOffset |
||||
*/ |
||||
class OffsetControlManager { |
||||
public static class Builder { |
||||
private LogContext logContext = null; |
||||
private SnapshotRegistry snapshotRegistry = null; |
||||
private QuorumControllerMetrics metrics = null; |
||||
private Time time = Time.SYSTEM; |
||||
|
||||
Builder setLogContext(LogContext logContext) { |
||||
this.logContext = logContext; |
||||
return this; |
||||
} |
||||
|
||||
Builder setSnapshotRegistry(SnapshotRegistry snapshotRegistry) { |
||||
this.snapshotRegistry = snapshotRegistry; |
||||
return this; |
||||
} |
||||
|
||||
Builder setMetrics(QuorumControllerMetrics metrics) { |
||||
this.metrics = metrics; |
||||
return this; |
||||
} |
||||
|
||||
Builder setTime(Time time) { |
||||
this.time = time; |
||||
return this; |
||||
} |
||||
|
||||
public OffsetControlManager build() { |
||||
if (logContext == null) logContext = new LogContext(); |
||||
if (snapshotRegistry == null) snapshotRegistry = new SnapshotRegistry(logContext); |
||||
if (metrics == null) { |
||||
metrics = new QuorumControllerMetrics(Optional.empty(), time, false); |
||||
} |
||||
return new OffsetControlManager(logContext, |
||||
snapshotRegistry, |
||||
metrics, |
||||
time); |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* The slf4j logger. |
||||
*/ |
||||
private final Logger log; |
||||
|
||||
/** |
||||
* The snapshot registry. |
||||
*/ |
||||
private final SnapshotRegistry snapshotRegistry; |
||||
|
||||
/** |
||||
* The quorum controller metrics. |
||||
*/ |
||||
private final QuorumControllerMetrics metrics; |
||||
|
||||
/** |
||||
* The clock. |
||||
*/ |
||||
private final Time time; |
||||
|
||||
/** |
||||
* The ID of the snapshot that we're currently replaying, or null if there is none. |
||||
*/ |
||||
private OffsetAndEpoch currentSnapshotId; |
||||
|
||||
/** |
||||
* The name of the snapshot that we're currently replaying, or null if there is none. |
||||
*/ |
||||
private String currentSnapshotName; |
||||
|
||||
/** |
||||
* The latest committed offset. |
||||
*/ |
||||
private long lastCommittedOffset; |
||||
|
||||
/** |
||||
* The latest committed epoch. |
||||
*/ |
||||
private int lastCommittedEpoch; |
||||
|
||||
/** |
||||
* The latest offset that it is safe to read from. |
||||
*/ |
||||
private long lastStableOffset; |
||||
|
||||
/** |
||||
* The offset of the transaction we're in, or -1 if we are not in one. |
||||
*/ |
||||
private long transactionStartOffset; |
||||
|
||||
/** |
||||
* The next offset we should write to, or -1 if the controller is not active. Exclusive offset. |
||||
*/ |
||||
private long nextWriteOffset; |
||||
|
||||
private OffsetControlManager( |
||||
LogContext logContext, |
||||
SnapshotRegistry snapshotRegistry, |
||||
QuorumControllerMetrics metrics, |
||||
Time time |
||||
) { |
||||
this.log = logContext.logger(OffsetControlManager.class); |
||||
this.snapshotRegistry = snapshotRegistry; |
||||
this.metrics = metrics; |
||||
this.time = time; |
||||
this.currentSnapshotId = null; |
||||
this.currentSnapshotName = null; |
||||
this.lastCommittedOffset = -1L; |
||||
this.lastCommittedEpoch = -1; |
||||
this.lastStableOffset = -1L; |
||||
this.transactionStartOffset = -1L; |
||||
this.nextWriteOffset = -1L; |
||||
snapshotRegistry.getOrCreateSnapshot(-1L); |
||||
metrics.setActive(false); |
||||
metrics.setLastCommittedRecordOffset(-1L); |
||||
metrics.setLastAppliedRecordOffset(-1L); |
||||
metrics.setLastAppliedRecordTimestamp(-1L); |
||||
} |
||||
|
||||
/** |
||||
* @return The SnapshotRegistry used by this offset control manager. |
||||
*/ |
||||
SnapshotRegistry snapshotRegistry() { |
||||
return snapshotRegistry; |
||||
} |
||||
|
||||
/** |
||||
* @return QuorumControllerMetrics managed by this offset control manager. |
||||
*/ |
||||
QuorumControllerMetrics metrics() { |
||||
return metrics; |
||||
} |
||||
|
||||
/** |
||||
* @return the ID of the current snapshot. |
||||
*/ |
||||
OffsetAndEpoch currentSnapshotId() { |
||||
return currentSnapshotId; |
||||
} |
||||
|
||||
/** |
||||
* @return the name of the current snapshot. |
||||
*/ |
||||
String currentSnapshotName() { |
||||
return currentSnapshotName; |
||||
} |
||||
|
||||
/** |
||||
* @return the last committed offset. |
||||
*/ |
||||
long lastCommittedOffset() { |
||||
return lastCommittedOffset; |
||||
} |
||||
|
||||
/** |
||||
* @return the last committed epoch. |
||||
*/ |
||||
int lastCommittedEpoch() { |
||||
return lastCommittedEpoch; |
||||
} |
||||
|
||||
/** |
||||
* @return the latest offset that it is safe to read from. |
||||
*/ |
||||
long lastStableOffset() { |
||||
return lastStableOffset; |
||||
} |
||||
|
||||
/** |
||||
* @return the transaction start offset, or -1 if there is no transaction. |
||||
*/ |
||||
long transactionStartOffset() { |
||||
return transactionStartOffset; |
||||
} |
||||
|
||||
/** |
||||
* @return the next offset that the active controller should write to. |
||||
*/ |
||||
long nextWriteOffset() { |
||||
return nextWriteOffset; |
||||
} |
||||
|
||||
/** |
||||
* @return true only if the manager is active. |
||||
*/ |
||||
boolean active() { |
||||
return nextWriteOffset != -1L; |
||||
} |
||||
|
||||
/** |
||||
* Called when the QuorumController becomes active. |
||||
* |
||||
* @param newNextWriteOffset The new next write offset to use. Must be non-negative. |
||||
*/ |
||||
void activate(long newNextWriteOffset) { |
||||
if (active()) { |
||||
throw new RuntimeException("Can't activate already active OffsetControlManager."); |
||||
} |
||||
if (newNextWriteOffset < 0) { |
||||
throw new RuntimeException("Invalid negative newNextWriteOffset " + |
||||
newNextWriteOffset + "."); |
||||
} |
||||
// Before switching to active, create an in-memory snapshot at the last committed
|
||||
// offset. This is required because the active controller assumes that there is always
|
||||
// an in-memory snapshot at the last committed offset.
|
||||
snapshotRegistry.getOrCreateSnapshot(lastStableOffset); |
||||
this.nextWriteOffset = newNextWriteOffset; |
||||
metrics.setActive(true); |
||||
} |
||||
|
||||
/** |
||||
* Called when the QuorumController becomes inactive. |
||||
*/ |
||||
void deactivate() { |
||||
if (!active()) { |
||||
throw new RuntimeException("Can't deactivate inactive OffsetControlManager."); |
||||
} |
||||
metrics.setActive(false); |
||||
metrics.setLastAppliedRecordOffset(lastStableOffset); |
||||
this.nextWriteOffset = -1L; |
||||
if (!snapshotRegistry.hasSnapshot(lastStableOffset)) { |
||||
throw new RuntimeException("Unable to reset to last stable offset " + lastStableOffset + |
||||
". No in-memory snapshot found for this offset."); |
||||
} |
||||
snapshotRegistry.revertToSnapshot(lastStableOffset); |
||||
} |
||||
|
||||
/** |
||||
* Handle the callback from the Raft layer indicating that a batch was committed. |
||||
* |
||||
* @param batch The batch that has been committed. |
||||
*/ |
||||
void handleCommitBatch(Batch<ApiMessageAndVersion> batch) { |
||||
this.lastCommittedOffset = batch.lastOffset(); |
||||
this.lastCommittedEpoch = batch.epoch(); |
||||
maybeAdvanceLastStableOffset(); |
||||
metrics.setLastCommittedRecordOffset(batch.lastOffset()); |
||||
if (!active()) { |
||||
// On standby controllers, the last applied record offset is equals to the last
|
||||
// committed offset.
|
||||
metrics.setLastAppliedRecordOffset(batch.lastOffset()); |
||||
metrics.setLastAppliedRecordTimestamp(batch.appendTimestamp()); |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* Called by the active controller after it has invoked scheduleAtomicAppend to schedule some |
||||
* records to be written. |
||||
* |
||||
* @param endOffset The offset of the last record that was written. |
||||
*/ |
||||
void handleScheduleAtomicAppend(long endOffset) { |
||||
this.nextWriteOffset = endOffset + 1; |
||||
|
||||
snapshotRegistry.getOrCreateSnapshot(endOffset); |
||||
|
||||
metrics.setLastAppliedRecordOffset(endOffset); |
||||
|
||||
// This is not truly the append timestamp. The KRaft client doesn't expose the append
|
||||
// time when scheduling a write. This is good enough because this is called right after
|
||||
// the records were given to the KRAft client for appending and the default append linger
|
||||
// for KRaft is 25ms.
|
||||
metrics.setLastAppliedRecordTimestamp(time.milliseconds()); |
||||
} |
||||
|
||||
/** |
||||
* Advance the last stable offset if needed. |
||||
*/ |
||||
void maybeAdvanceLastStableOffset() { |
||||
long newLastStableOffset; |
||||
if (transactionStartOffset == -1L) { |
||||
newLastStableOffset = lastCommittedOffset; |
||||
} else { |
||||
newLastStableOffset = Math.min(transactionStartOffset - 1, lastCommittedOffset); |
||||
} |
||||
if (lastStableOffset < newLastStableOffset) { |
||||
lastStableOffset = newLastStableOffset; |
||||
snapshotRegistry.deleteSnapshotsUpTo(lastStableOffset); |
||||
if (!active()) { |
||||
snapshotRegistry.getOrCreateSnapshot(lastStableOffset); |
||||
} |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* Called before we load a Raft snapshot. |
||||
* |
||||
* @param snapshotId The Raft snapshot offset and epoch. |
||||
*/ |
||||
void beginLoadSnapshot(OffsetAndEpoch snapshotId) { |
||||
if (currentSnapshotId != null) { |
||||
throw new RuntimeException("Can't begin reading snapshot for " + snapshotId + |
||||
", because we are already reading " + currentSnapshotId); |
||||
} |
||||
this.currentSnapshotId = snapshotId; |
||||
this.currentSnapshotName = Snapshots.filenameFromSnapshotId(snapshotId); |
||||
log.info("Starting to load snapshot {}. Previous lastCommittedOffset was {}. Previous " + |
||||
"transactionStartOffset was {}.", currentSnapshotName, lastCommittedOffset, |
||||
transactionStartOffset); |
||||
this.snapshotRegistry.reset(); |
||||
this.lastCommittedOffset = -1L; |
||||
this.lastCommittedEpoch = -1; |
||||
this.lastStableOffset = -1L; |
||||
this.transactionStartOffset = -1L; |
||||
this.nextWriteOffset = -1L; |
||||
} |
||||
|
||||
/** |
||||
* Called after we have finished loading a Raft snapshot. |
||||
* |
||||
* @param timestamp The timestamp of the snapshot. |
||||
*/ |
||||
void endLoadSnapshot(long timestamp) { |
||||
if (currentSnapshotId == null) { |
||||
throw new RuntimeException("Can't end loading snapshot, because there is no " + |
||||
"current snapshot."); |
||||
} |
||||
log.info("Successfully loaded snapshot {}.", currentSnapshotName); |
||||
this.snapshotRegistry.getOrCreateSnapshot(currentSnapshotId.offset()); |
||||
this.lastCommittedOffset = currentSnapshotId.offset(); |
||||
this.lastCommittedEpoch = currentSnapshotId.epoch(); |
||||
this.lastStableOffset = currentSnapshotId.offset(); |
||||
this.transactionStartOffset = -1L; |
||||
this.nextWriteOffset = -1L; |
||||
metrics.setLastCommittedRecordOffset(currentSnapshotId.offset()); |
||||
metrics.setLastAppliedRecordOffset(currentSnapshotId.offset()); |
||||
metrics.setLastAppliedRecordTimestamp(timestamp); |
||||
this.currentSnapshotId = null; |
||||
this.currentSnapshotName = null; |
||||
} |
||||
|
||||
public void replay(BeginTransactionRecord message, long offset) { |
||||
if (currentSnapshotId != null) { |
||||
throw new RuntimeException("BeginTransactionRecord cannot appear within a snapshot."); |
||||
} |
||||
if (transactionStartOffset != -1L) { |
||||
throw new RuntimeException("Can't replay a BeginTransactionRecord at " + offset + |
||||
" because the transaction at " + transactionStartOffset + " was never closed."); |
||||
} |
||||
snapshotRegistry.getOrCreateSnapshot(offset - 1); |
||||
transactionStartOffset = offset; |
||||
log.info("Replayed {} at offset {}.", message, offset); |
||||
} |
||||
|
||||
public void replay(EndTransactionRecord message, long offset) { |
||||
if (currentSnapshotId != null) { |
||||
throw new RuntimeException("EndTransactionRecord cannot appear within a snapshot."); |
||||
} |
||||
if (transactionStartOffset == -1L) { |
||||
throw new RuntimeException("Can't replay an EndTransactionRecord at " + offset + |
||||
" because there is no open transaction."); |
||||
} |
||||
transactionStartOffset = -1L; |
||||
log.info("Replayed {} at offset {}.", message, offset); |
||||
} |
||||
|
||||
public void replay(AbortTransactionRecord message, long offset) { |
||||
if (currentSnapshotId != null) { |
||||
throw new RuntimeException("AbortTransactionRecord cannot appear within a snapshot."); |
||||
} |
||||
if (transactionStartOffset == -1L) { |
||||
throw new RuntimeException("Can't replay an AbortTransactionRecord at " + offset + |
||||
" because there is no open transaction."); |
||||
} |
||||
long preTransactionOffset = transactionStartOffset - 1; |
||||
snapshotRegistry.revertToSnapshot(preTransactionOffset); |
||||
transactionStartOffset = -1L; |
||||
log.info("Replayed {} at offset {}. Reverted to offset {}.", |
||||
message, offset, preTransactionOffset); |
||||
} |
||||
|
||||
// VisibleForTesting
|
||||
void setNextWriteOffset(long newNextWriteOffset) { |
||||
this.nextWriteOffset = newNextWriteOffset; |
||||
} |
||||
} |
||||
@ -0,0 +1,27 @@
@@ -0,0 +1,27 @@
|
||||
// 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. |
||||
|
||||
{ |
||||
"apiKey": 25, |
||||
"type": "metadata", |
||||
"name": "AbortTransactionRecord", |
||||
"validVersions": "0", |
||||
"flexibleVersions": "0+", |
||||
"fields": [ |
||||
{ "name": "Reason", "type": "string", "default": "null", |
||||
"versions": "0+", "nullableVersions": "0+", "taggedVersions": "0+", "tag": 0, |
||||
"about": "An optional textual description of why the transaction was aborted." } |
||||
] |
||||
} |
||||
@ -0,0 +1,27 @@
@@ -0,0 +1,27 @@
|
||||
// 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. |
||||
|
||||
{ |
||||
"apiKey": 23, |
||||
"type": "metadata", |
||||
"name": "BeginTransactionRecord", |
||||
"validVersions": "0", |
||||
"flexibleVersions": "0+", |
||||
"fields": [ |
||||
{ "name": "Name", "type": "string", "default": "null", |
||||
"versions": "0+", "nullableVersions": "0+", "taggedVersions": "0+", "tag": 0, |
||||
"about": "An optional textual description of this transaction." } |
||||
] |
||||
} |
||||
@ -0,0 +1,24 @@
@@ -0,0 +1,24 @@
|
||||
// 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. |
||||
|
||||
{ |
||||
"apiKey": 24, |
||||
"type": "metadata", |
||||
"name": "EndTransactionRecord", |
||||
"validVersions": "0", |
||||
"flexibleVersions": "0+", |
||||
"fields": [ |
||||
] |
||||
} |
||||
@ -0,0 +1,270 @@
@@ -0,0 +1,270 @@
|
||||
/* |
||||
* 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.controller; |
||||
|
||||
import org.apache.kafka.common.metadata.AbortTransactionRecord; |
||||
import org.apache.kafka.common.metadata.BeginTransactionRecord; |
||||
import org.apache.kafka.common.metadata.EndTransactionRecord; |
||||
import org.apache.kafka.common.metadata.NoOpRecord; |
||||
import org.apache.kafka.common.utils.LogContext; |
||||
import org.apache.kafka.raft.Batch; |
||||
import org.apache.kafka.raft.OffsetAndEpoch; |
||||
import org.apache.kafka.server.common.ApiMessageAndVersion; |
||||
import org.apache.kafka.timeline.TrackingSnapshotRegistry; |
||||
import org.junit.jupiter.api.Test; |
||||
import org.junit.jupiter.api.Timeout; |
||||
import org.junit.jupiter.params.ParameterizedTest; |
||||
import org.junit.jupiter.params.provider.ValueSource; |
||||
|
||||
import java.util.Arrays; |
||||
import java.util.Collections; |
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals; |
||||
import static org.junit.jupiter.api.Assertions.assertFalse; |
||||
import static org.junit.jupiter.api.Assertions.assertNull; |
||||
import static org.junit.jupiter.api.Assertions.assertThrows; |
||||
import static org.junit.jupiter.api.Assertions.assertTrue; |
||||
|
||||
|
||||
@Timeout(value = 40) |
||||
public class OffsetControlManagerTest { |
||||
@Test |
||||
public void testInitialValues() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
assertNull(offsetControl.currentSnapshotId()); |
||||
assertNull(offsetControl.currentSnapshotName()); |
||||
assertEquals(-1L, offsetControl.lastCommittedOffset()); |
||||
assertEquals(-1, offsetControl.lastCommittedEpoch()); |
||||
assertEquals(-1, offsetControl.lastStableOffset()); |
||||
assertEquals(-1, offsetControl.transactionStartOffset()); |
||||
assertEquals(-1, offsetControl.nextWriteOffset()); |
||||
assertFalse(offsetControl.active()); |
||||
assertEquals(Arrays.asList(-1L), offsetControl.snapshotRegistry().epochsList()); |
||||
} |
||||
|
||||
@Test |
||||
public void testActivate() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
offsetControl.activate(1000L); |
||||
assertEquals(1000L, offsetControl.nextWriteOffset()); |
||||
assertTrue(offsetControl.active()); |
||||
assertTrue(offsetControl.metrics().active()); |
||||
assertEquals(Arrays.asList(-1L), offsetControl.snapshotRegistry().epochsList()); |
||||
} |
||||
|
||||
@Test |
||||
public void testActivateFailsIfAlreadyActive() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
offsetControl.activate(1000L); |
||||
assertEquals("Can't activate already active OffsetControlManager.", |
||||
assertThrows(RuntimeException.class, |
||||
() -> offsetControl.activate(2000L)). |
||||
getMessage()); |
||||
} |
||||
|
||||
@Test |
||||
public void testActivateFailsIfNewNextWriteOffsetIsNegative() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
assertEquals("Invalid negative newNextWriteOffset -2.", |
||||
assertThrows(RuntimeException.class, |
||||
() -> offsetControl.activate(-2)). |
||||
getMessage()); |
||||
} |
||||
|
||||
@Test |
||||
public void testActivateAndDeactivate() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
offsetControl.activate(1000L); |
||||
assertEquals(1000L, offsetControl.nextWriteOffset()); |
||||
offsetControl.deactivate(); |
||||
assertEquals(-1L, offsetControl.nextWriteOffset()); |
||||
} |
||||
|
||||
@Test |
||||
public void testDeactivateFailsIfNotActive() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
assertEquals("Can't deactivate inactive OffsetControlManager.", |
||||
assertThrows(RuntimeException.class, |
||||
() -> offsetControl.deactivate()). |
||||
getMessage()); |
||||
} |
||||
|
||||
private static Batch<ApiMessageAndVersion> newFakeBatch( |
||||
long lastOffset, |
||||
int epoch, |
||||
long appendTimestamp |
||||
) { |
||||
return Batch.data( |
||||
lastOffset, |
||||
epoch, |
||||
appendTimestamp, |
||||
100, |
||||
Collections.singletonList(new ApiMessageAndVersion(new NoOpRecord(), (short) 0))); |
||||
} |
||||
|
||||
@Test |
||||
public void testHandleCommitBatch() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
|
||||
offsetControl.handleCommitBatch(newFakeBatch(1000L, 200, 3000L)); |
||||
assertEquals(Arrays.asList(1000L), offsetControl.snapshotRegistry().epochsList()); |
||||
assertEquals(1000L, offsetControl.lastCommittedOffset()); |
||||
assertEquals(200, offsetControl.lastCommittedEpoch()); |
||||
assertEquals(1000L, offsetControl.lastStableOffset()); |
||||
assertEquals(-1L, offsetControl.transactionStartOffset()); |
||||
assertEquals(-1L, offsetControl.nextWriteOffset()); |
||||
assertFalse(offsetControl.active()); |
||||
assertFalse(offsetControl.metrics().active()); |
||||
assertEquals(1000L, offsetControl.metrics().lastAppliedRecordOffset()); |
||||
assertEquals(1000L, offsetControl.metrics().lastCommittedRecordOffset()); |
||||
assertEquals(3000L, offsetControl.metrics().lastAppliedRecordTimestamp()); |
||||
} |
||||
|
||||
@Test |
||||
public void testHandleScheduleAtomicAppend() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
|
||||
offsetControl.handleScheduleAtomicAppend(2000L); |
||||
assertEquals(2001L, offsetControl.nextWriteOffset()); |
||||
assertEquals(2000L, offsetControl.metrics().lastAppliedRecordOffset()); |
||||
assertEquals(-1L, offsetControl.lastStableOffset()); |
||||
assertEquals(-1L, offsetControl.lastCommittedOffset()); |
||||
assertEquals(Arrays.asList(-1L, 2000L), offsetControl.snapshotRegistry().epochsList()); |
||||
|
||||
offsetControl.handleCommitBatch(newFakeBatch(2000L, 200, 3000L)); |
||||
assertEquals(2000L, offsetControl.lastStableOffset()); |
||||
assertEquals(2000L, offsetControl.lastCommittedOffset()); |
||||
assertEquals(Arrays.asList(2000L), offsetControl.snapshotRegistry().epochsList()); |
||||
} |
||||
|
||||
@Test |
||||
public void testHandleLoadSnapshot() { |
||||
TrackingSnapshotRegistry snapshotRegistry = new TrackingSnapshotRegistry(new LogContext()); |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder(). |
||||
setSnapshotRegistry(snapshotRegistry). |
||||
build(); |
||||
|
||||
offsetControl.beginLoadSnapshot(new OffsetAndEpoch(4000L, 300)); |
||||
assertEquals(Arrays.asList("snapshot[-1]", "reset"), snapshotRegistry.operations()); |
||||
assertEquals(new OffsetAndEpoch(4000L, 300), offsetControl.currentSnapshotId()); |
||||
assertEquals("00000000000000004000-0000000300", offsetControl.currentSnapshotName()); |
||||
assertEquals(Arrays.asList(), offsetControl.snapshotRegistry().epochsList()); |
||||
|
||||
offsetControl.endLoadSnapshot(3456L); |
||||
assertEquals(Arrays.asList("snapshot[-1]", "reset", "snapshot[4000]"), |
||||
snapshotRegistry.operations()); |
||||
assertNull(offsetControl.currentSnapshotId()); |
||||
assertNull(offsetControl.currentSnapshotName()); |
||||
assertEquals(Arrays.asList(4000L), offsetControl.snapshotRegistry().epochsList()); |
||||
assertEquals(4000L, offsetControl.lastCommittedOffset()); |
||||
assertEquals(300, offsetControl.lastCommittedEpoch()); |
||||
assertEquals(4000L, offsetControl.lastStableOffset()); |
||||
assertEquals(-1L, offsetControl.transactionStartOffset()); |
||||
assertEquals(-1L, offsetControl.nextWriteOffset()); |
||||
assertEquals(4000L, offsetControl.metrics().lastCommittedRecordOffset()); |
||||
assertEquals(4000L, offsetControl.metrics().lastAppliedRecordOffset()); |
||||
assertEquals(3456L, offsetControl.metrics().lastAppliedRecordTimestamp()); |
||||
} |
||||
|
||||
@Test |
||||
public void testBeginTransactionRecordNotAllowedInSnapshot() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
offsetControl.beginLoadSnapshot(new OffsetAndEpoch(4000L, 300)); |
||||
assertEquals("BeginTransactionRecord cannot appear within a snapshot.", |
||||
assertThrows(RuntimeException.class, |
||||
() -> offsetControl.replay(new BeginTransactionRecord(), 1000L)). |
||||
getMessage()); |
||||
} |
||||
|
||||
@Test |
||||
public void testEndTransactionRecordNotAllowedInSnapshot() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
offsetControl.beginLoadSnapshot(new OffsetAndEpoch(4000L, 300)); |
||||
assertEquals("EndTransactionRecord cannot appear within a snapshot.", |
||||
assertThrows(RuntimeException.class, |
||||
() -> offsetControl.replay(new EndTransactionRecord(), 1000L)). |
||||
getMessage()); |
||||
} |
||||
|
||||
@Test |
||||
public void testAbortTransactionRecordNotAllowedInSnapshot() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
offsetControl.beginLoadSnapshot(new OffsetAndEpoch(4000L, 300)); |
||||
assertEquals("AbortTransactionRecord cannot appear within a snapshot.", |
||||
assertThrows(RuntimeException.class, |
||||
() -> offsetControl.replay(new AbortTransactionRecord(), 1000L)). |
||||
getMessage()); |
||||
} |
||||
|
||||
@Test |
||||
public void testEndLoadSnapshotFailsWhenNotInSnapshot() { |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder().build(); |
||||
assertEquals("Can't end loading snapshot, because there is no current snapshot.", |
||||
assertThrows(RuntimeException.class, |
||||
() -> offsetControl.endLoadSnapshot(1000L)). |
||||
getMessage()); |
||||
} |
||||
|
||||
@ParameterizedTest |
||||
@ValueSource(booleans = {false, true}) |
||||
public void testReplayTransaction(boolean aborted) { |
||||
TrackingSnapshotRegistry snapshotRegistry = new TrackingSnapshotRegistry(new LogContext()); |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder(). |
||||
setSnapshotRegistry(snapshotRegistry). |
||||
build(); |
||||
|
||||
offsetControl.replay(new BeginTransactionRecord(), 1500L); |
||||
assertEquals(1500L, offsetControl.transactionStartOffset()); |
||||
assertEquals(Arrays.asList(-1L, 1499L), offsetControl.snapshotRegistry().epochsList()); |
||||
|
||||
offsetControl.handleCommitBatch(newFakeBatch(1550L, 100, 2000L)); |
||||
assertEquals(1550L, offsetControl.lastCommittedOffset()); |
||||
assertEquals(100, offsetControl.lastCommittedEpoch()); |
||||
assertEquals(1499L, offsetControl.lastStableOffset()); |
||||
assertEquals(Arrays.asList(1499L), offsetControl.snapshotRegistry().epochsList()); |
||||
|
||||
if (aborted) { |
||||
offsetControl.replay(new AbortTransactionRecord(), 1600L); |
||||
assertEquals(Arrays.asList("snapshot[-1]", "snapshot[1499]", "revert[1499]"), |
||||
snapshotRegistry.operations()); |
||||
} else { |
||||
offsetControl.replay(new EndTransactionRecord(), 1600L); |
||||
assertEquals(Arrays.asList("snapshot[-1]", "snapshot[1499]"), |
||||
snapshotRegistry.operations()); |
||||
} |
||||
assertEquals(-1L, offsetControl.transactionStartOffset()); |
||||
assertEquals(1499L, offsetControl.lastStableOffset()); |
||||
|
||||
offsetControl.handleCommitBatch(newFakeBatch(1650, 100, 2100L)); |
||||
assertEquals(1650, offsetControl.lastStableOffset()); |
||||
assertEquals(Arrays.asList(1650L), offsetControl.snapshotRegistry().epochsList()); |
||||
} |
||||
|
||||
@Test |
||||
public void testLoadSnapshotClearsTransactionalState() { |
||||
TrackingSnapshotRegistry snapshotRegistry = new TrackingSnapshotRegistry(new LogContext()); |
||||
OffsetControlManager offsetControl = new OffsetControlManager.Builder(). |
||||
setSnapshotRegistry(snapshotRegistry). |
||||
build(); |
||||
offsetControl.replay(new BeginTransactionRecord(), 1500L); |
||||
offsetControl.beginLoadSnapshot(new OffsetAndEpoch(4000L, 300)); |
||||
assertEquals(-1L, offsetControl.transactionStartOffset()); |
||||
assertEquals(Arrays.asList("snapshot[-1]", "snapshot[1499]", "reset"), |
||||
snapshotRegistry.operations()); |
||||
} |
||||
} |
||||
@ -0,0 +1,56 @@
@@ -0,0 +1,56 @@
|
||||
/* |
||||
* 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.timeline; |
||||
|
||||
import org.apache.kafka.common.utils.LogContext; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.List; |
||||
|
||||
|
||||
public class TrackingSnapshotRegistry extends SnapshotRegistry { |
||||
private final List<String> operations = new ArrayList<>(); |
||||
|
||||
public TrackingSnapshotRegistry(LogContext logContext) { |
||||
super(logContext); |
||||
} |
||||
|
||||
public List<String> operations() { |
||||
return new ArrayList<>(operations); |
||||
} |
||||
|
||||
@Override |
||||
public void revertToSnapshot(long targetEpoch) { |
||||
operations.add("revert[" + targetEpoch + "]"); |
||||
super.revertToSnapshot(targetEpoch); |
||||
} |
||||
|
||||
@Override |
||||
public void reset() { |
||||
operations.add("reset"); |
||||
super.reset(); |
||||
} |
||||
|
||||
@Override |
||||
public Snapshot getOrCreateSnapshot(long epoch) { |
||||
if (!hasSnapshot(epoch)) { |
||||
operations.add("snapshot[" + epoch + "]"); |
||||
} |
||||
return super.getOrCreateSnapshot(epoch); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue