This is an automated email from the ASF dual-hosted git repository. bdeggleston pushed a commit to branch cep-45-mutation-tracking in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 2ef5f93aa0edd4ddf669894e268ba4acbe39d86f Author: Blake Eggleston <[email protected]> AuthorDate: Tue Jun 2 14:23:23 2026 -0700 CEP-45: Miscellaneous mutation tracking bug and test fixes - Handle dropped tables during mutation journal recovery - Track journal segment references via coordinator log offset overlap - Fix dtest flakiness and incorrect assertions across mutation tracking tests - Explicitly configure memtable and compaction strategies in affected unit tests Patch by Blake Eggleston; Reviewed by Francisco Guerrero for CASSANDRA-21563 --- .../replication/CoordinatorLogOffsets.java | 2 + .../ImmutableCoordinatorLogOffsets.java | 9 +++ .../cassandra/replication/MutationJournal.java | 29 +++++++ .../replication/UnreconciledMutations.java | 18 ++++- .../repair/MutationTrackingReadRepairTest.java | 2 +- .../test/repair/MutationTrackingRepairTest.java | 5 +- .../MutationTrackingSyncCoordinatorTest.java | 15 ++-- .../distributed/test/sai/IndexStreamingTest.java | 2 +- .../PaxosMutationTrackingForwardingV1Test.java | 50 ++++++------ .../test/tracking/TrackedImportFailureTest.java | 2 +- .../test/tracking/TrackedRepairFailureTest.java | 2 +- .../TrackedUnreconciledPromotionTest.java | 1 + .../db/memtable/SplitDomainMemtableTest.java | 5 ++ .../cassandra/replication/CoordinatorLogTest.java | 14 ++++ .../replication/UnreconciledMutationsTest.java | 92 ++++++++++++++++++++++ 15 files changed, 204 insertions(+), 44 deletions(-) diff --git a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java index 5a394fc08d..213413456d 100644 --- a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java +++ b/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java @@ -47,6 +47,8 @@ public interface CoordinatorLogOffsets<O extends Offsets> Mutations<O> mutations(); + boolean isEmpty(); + default ActivatedTransfers transfers() { return ActivatedTransfers.EMPTY; diff --git a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java index d2de41ca8b..5ebf038923 100644 --- a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java +++ b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java @@ -29,6 +29,7 @@ import java.util.function.Predicate; import javax.annotation.concurrent.NotThreadSafe; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterators; @@ -160,6 +161,14 @@ public class ImmutableCoordinatorLogOffsets implements CoordinatorLogOffsets<Off return this; } + @VisibleForTesting + Builder add(long logId, int start, int end) + { + ids.computeIfAbsent(logId, k -> new Offsets.Immutable.Builder(new CoordinatorLogId(k))) + .add(start, end); + return this; + } + private Builder addAll(CoordinatorLogOffsets.Mutations<? extends Offsets> mutations) { for (long log : mutations) diff --git a/src/java/org/apache/cassandra/replication/MutationJournal.java b/src/java/org/apache/cassandra/replication/MutationJournal.java index 22a66f61c6..8758588bd4 100644 --- a/src/java/org/apache/cassandra/replication/MutationJournal.java +++ b/src/java/org/apache/cassandra/replication/MutationJournal.java @@ -53,6 +53,7 @@ import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.db.commitlog.CommitLogPosition; import org.apache.cassandra.db.partitions.PartitionUpdate; +import org.apache.cassandra.exceptions.UnknownTableException; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.io.util.File; @@ -427,6 +428,25 @@ public class MutationJournal { replay(new DeserializedRecordConsumer<>(MutationSerializer.INSTANCE) { + @Override + public void accept(long segment, int position, ShortMutationId key, ByteBuffer buffer, int userVersion) + { + try + { + super.accept(segment, position, key, buffer, userVersion); + } + catch (RuntimeException e) + { + if (e.getCause() instanceof UnknownTableException) + { + UnknownTableException ute = (UnknownTableException) e.getCause(); + logger.warn("Skipping replaying mutation {} from unknown (probably removed) table with id {}", key, ute.id); + return; + } + throw e; + } + } + @Override protected void accept(long segmentId, int position, ShortMutationId key, Mutation value) { @@ -943,6 +963,15 @@ public class MutationJournal * Test helpers */ + @VisibleForTesting + void truncateForTesting() + { + journal.truncateForTesting(); + segmentStateTrackers.clear(); + pendingClearReplay.clear(); + lastSegmentTracker = null; + } + @VisibleForTesting public void closeCurrentSegmentForTestingIfNonEmpty() { diff --git a/src/java/org/apache/cassandra/replication/UnreconciledMutations.java b/src/java/org/apache/cassandra/replication/UnreconciledMutations.java index b78e84d256..81778518bd 100644 --- a/src/java/org/apache/cassandra/replication/UnreconciledMutations.java +++ b/src/java/org/apache/cassandra/replication/UnreconciledMutations.java @@ -36,6 +36,7 @@ import org.apache.cassandra.db.PartitionPosition; import org.apache.cassandra.dht.AbstractBounds; import org.apache.cassandra.dht.Bounds; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.exceptions.UnknownTableException; import org.apache.cassandra.schema.TableId; /** @@ -268,7 +269,22 @@ public class UnreconciledMutations for (int offset = iter.start(), end = iter.end(); offset <= end; offset++) { ShortMutationId id = new ShortMutationId(witnessed.logId, offset); - Mutation mutation = MutationJournal.instance().read(id); + Mutation mutation; + try + { + mutation = MutationJournal.instance().read(id); + } + catch (RuntimeException e) + { + if (e.getCause() instanceof UnknownTableException) + { + UnknownTableException ute = (UnknownTableException) e.getCause(); + logger.warn("Skipping loading mutation {} from unknown (probably removed) table with id {}", id, ute.id); + continue; + } + throw e; + } + if (mutation != null) { result.addDirectly(mutation); diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java index 7134a9d5cd..bf4a437f6d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java @@ -104,7 +104,7 @@ public class MutationTrackingReadRepairTest extends TestBaseImpl .set("repair.retries.max_attempts", 10) .set("repair.retries.base_sleep_time", "100ms") .set("repair.retries.max_sleep_time", "500ms") - .with(Feature.GOSSIP)) + .with(Feature.GOSSIP, Feature.NETWORK)) .start(); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java index f5ad9f7d26..87b1eb0457 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java @@ -881,11 +881,12 @@ public class MutationTrackingRepairTest extends TestBaseImpl @Test public void testRepairFailsOnTopologyChange() throws Exception { - insertData("tbl", 0, 50); - // Block offset broadcasts so the sync coordinator stays alive waiting IMessageFilters.Filter offsetFilter = CLUSTER.filters().verbs(Verb.MT_BROADCAST_LOG_OFFSETS.id).drop(); + // must use inconsistent data to prevent write process from marking the writes reconciled + insertDataWithInconsistency("tbl", 0, 50); + // Use a latch to detect when the sync request has been sent, meaning // the sync coordinator is active and tracking shard references CountDownLatch syncStarted = new CountDownLatch(1); diff --git a/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java b/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java index 7a13f606f2..c1d82ef104 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java @@ -29,6 +29,7 @@ import org.apache.cassandra.dht.Token; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.net.Verb; import org.apache.cassandra.repair.RepairJobDesc; import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.replication.MutationTrackingService; @@ -64,12 +65,6 @@ public class MutationTrackingSyncCoordinatorTest extends TestBaseImpl return KS_NAME + suffix + '.' + TBL_NAME; } - private void pauseOffsetBroadcasts(Cluster cluster, boolean pause) - { - for (int i = 1; i <= cluster.size(); i++) - cluster.get(i).runOnInstance(() -> MutationTrackingService.instance().pauseOffsetBroadcast(pause)); - } - private static Range<Token> fullTokenRange() { return new Range<>( @@ -233,9 +228,6 @@ public class MutationTrackingSyncCoordinatorTest extends TestBaseImpl { createTrackedKeyspace(cluster, "4"); - // Pause offset broadcasts on all nodes to prevent sync from completing - pauseOffsetBroadcasts(cluster, true); - for (int i = 0; i < 100; i++) { cluster.coordinator(1).execute( @@ -243,7 +235,10 @@ public class MutationTrackingSyncCoordinatorTest extends TestBaseImpl ConsistencyLevel.ONE, i, i); } - // Start coordinator - it will be stuck waiting for offsets + // Drop sync responses so the coordinator never receives them + cluster.filters().verbs(Verb.MT_SYNC_RSP.id).drop(); + + // Start coordinator - it will be stuck waiting for sync responses Boolean wasCancelled = cluster.get(1).callOnInstance(() -> { Range<Token> range = fullTokenRange(); RepairJobDesc desc = new RepairJobDesc(TimeUUID.Generator.nextTimeUUID(), diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java index db3d841bdd..ebec55383b 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java @@ -69,7 +69,7 @@ public class IndexStreamingTest extends TestBaseImpl @Parameterized.Parameter(2) public boolean isWide; - @Parameterized.Parameters(name = "isLiteral={0}, isZeroCopyStreaming={1}") + @Parameterized.Parameters(name = "isLiteral={0}, isZeroCopyStreaming={1}, isWide={2}") public static List<Object[]> data() { List<Object[]> result = new ArrayList<>(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/PaxosMutationTrackingForwardingV1Test.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/PaxosMutationTrackingForwardingV1Test.java index f88f074476..d059aaa941 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/PaxosMutationTrackingForwardingV1Test.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/PaxosMutationTrackingForwardingV1Test.java @@ -215,32 +215,29 @@ public class PaxosMutationTrackingForwardingV1Test extends TestBaseImpl cluster.coordinator(1).execute("INSERT INTO " + ks + ".tbl (k, v) VALUES (" + KEY + ", 1)", ConsistencyLevel.ALL); - // Strategy: Hold PAXOS_COMMIT_REQ at nodes 2 and 3 (the non-forwarding replicas). - // While held, ALTER the keyspace to untracked. When released, nodes 2,3 see the - // commit request from the stale forward handler (node 1) and reject with - // COORDINATOR_BEHIND. This bubbles back to node 4's retry loop. - AssertingLatch commitArrived = new AssertingLatch("PAXOS_COMMIT_REQ at nodes 2,3"); + AtomicInteger forwardTarget = new AtomicInteger(-1); + AssertingLatch commitArrived = new AssertingLatch("PAXOS_COMMIT_REQ from forward target"); AssertingLatch alterDone = new AssertingLatch("V1 commit forwarding retry - alter to untracked"); - - // Count ALL inbound PAXOS_COMMIT_REQ at nodes 2,3 regardless of source. After the retry, - // node 4 (now on untracked) calls commitPaxosUntracked directly and sends to replicas - // {1,2,3}, yielding 2 more messages at {2,3}. Total = initial 2 + retry 2 = 4. - // If the retry didn't happen (regression), total would be 2 and the CAS would fail. AtomicInteger commitsAtReplicas = new AtomicInteger(); - - // Hold PAXOS_COMMIT_REQ at nodes 2 and 3 until ALTER completes. - // Also check retry messages (from node 4) for mutation ID absence. One filter handles - // both roles because the from-node distinguishes initial (node 1) from retry (node 4) - // and two overlapping MessageSpy instances on the same verb+destination would not - // compose cleanly — the cluster filter system applies ALL matching filters. AtomicInteger retryCommitsWithId = new AtomicInteger(); + + // Learn which replica the proximity measures chose for the forward — any of {1,2,3} is possible. + cluster.filters() + .verbs(Verb.PAXOS_COMMIT_FORWARD_REQ.id) + .messagesMatching((from, to, msg) -> { + forwardTarget.set(to); + return false; + }).drop(); + + // Hold PAXOS_COMMIT_REQ from the forward target until ALTER completes. Count ALL inbound + // PAXOS_COMMIT_REQ at replicas to verify the retry fires, and check retry messages (from + // node 4) for mutation ID absence. Conditional behaviour inside one filter — stays manual. cluster.filters() .verbs(Verb.PAXOS_COMMIT_REQ.id) - .to(2, 3) + .to(1, 2, 3) .messagesMatching((from, to, msg) -> { commitsAtReplicas.incrementAndGet(); - // Only hold commits from node 1 (the forward handler acting as coordinator) - if (from == 1) + if (from == forwardTarget.get()) { commitArrived.countDown(); alterDone.await(); @@ -269,8 +266,8 @@ public class PaxosMutationTrackingForwardingV1Test extends TestBaseImpl } finally { - // Release the held commits — nodes 2,3 now see migration mismatch -> COORDINATOR_BEHIND - alterDone.countDown(); + alterDone.release(); + commitArrived.release(); } // With the fix: node 4's retry loop catches CoordinatorBehindException, retries @@ -291,12 +288,11 @@ public class PaxosMutationTrackingForwardingV1Test extends TestBaseImpl // (node 1 gets it via the tracked local write in the initial forward, nodes 2,3 via retry). assertReplicasHaveValue(cluster, ks, KEY, 100, 1, 2, 3); - // Authenticate that the retry actually happened by counting inbound PAXOS_COMMIT_REQ at - // the two non-forwarding replicas. Initial attempt (from node 1, held then released) = 2, - // retry (from node 4 directly on untracked path) = 2. Total = 4. A result of 2 would mean - // the CAS succeeded without retry (bug). - assertEquals("Expected initial (2) + retry (2) = 4 PAXOS_COMMIT_REQ at nodes 2,3", - 4, commitsAtReplicas.get()); + // Retry verification: forward target sends to 2 other replicas (initial attempt), node 4 + // retries directly on the untracked path sending to all 3 replicas. Total = 5. + assertTrue("Forward target should have been identified", forwardTarget.get() > 0); + assertEquals("Expected initial (2) + retry (3) = 5 PAXOS_COMMIT_REQ at replicas", + 5, commitsAtReplicas.get()); // Retry messages from node 4 (untracked path) must not carry mutation IDs assertEquals("Retry PAXOS_COMMIT_REQ from node 4 should NOT carry mutation IDs (untracked path)", diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java index 623013ce10..7973e1037d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java @@ -73,7 +73,7 @@ public class TrackedImportFailureTest extends TrackedTransferTestBase public void importMissedActivation(ActivationRequest.Phase phase) throws Throwable { int MISSED_ACTIVATION = 2; - try (Cluster cluster = cluster(TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.install(MISSED_ACTIVATION))) + try (Cluster cluster = disableBackgroundReconciler(cluster(TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.install(MISSED_ACTIVATION)))) { TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.setup(cluster, phase); createSchema(cluster); diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedRepairFailureTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedRepairFailureTest.java index 8a49fcb4de..dca7695e61 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedRepairFailureTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedRepairFailureTest.java @@ -340,7 +340,7 @@ public class TrackedRepairFailureTest extends TrackedRepairTransferTestBase IInvokableInstance coordinator = cluster.get(1); IInvokableInstance receiver = cluster.get(2); - coordinator.nodetoolResult("repair", "--full", keyspace).asserts().success(); + coordinator.nodetoolResult("repair", keyspace).asserts().success(); coordinator.executeInternal("INSERT INTO " + tableWithKeyspace(keyspace) + " (pk, v) VALUES (?, 1)", KEY_100); coordinator.flush(keyspace); diff --git a/test/unit/org/apache/cassandra/db/compaction/TrackedUnreconciledPromotionTest.java b/test/unit/org/apache/cassandra/db/compaction/TrackedUnreconciledPromotionTest.java index 284e644128..7d36ffdc23 100644 --- a/test/unit/org/apache/cassandra/db/compaction/TrackedUnreconciledPromotionTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/TrackedUnreconciledPromotionTest.java @@ -122,6 +122,7 @@ public class TrackedUnreconciledPromotionTest TableMetadata.builder(ks, "tbl") .addPartitionKeyColumn("k", Int32Type.instance) .addRegularColumn("v", Int32Type.instance) + .compaction(CompactionParams.stcs(Collections.emptyMap())) .build(); SchemaLoader.createKeyspace(ks, KeyspaceParams.simple(1, ReplicationType.tracked), tableMetadata); diff --git a/test/unit/org/apache/cassandra/db/memtable/SplitDomainMemtableTest.java b/test/unit/org/apache/cassandra/db/memtable/SplitDomainMemtableTest.java index b8d8ff1c72..1ca389da74 100644 --- a/test/unit/org/apache/cassandra/db/memtable/SplitDomainMemtableTest.java +++ b/test/unit/org/apache/cassandra/db/memtable/SplitDomainMemtableTest.java @@ -62,8 +62,10 @@ import org.apache.cassandra.replication.MutationId; import org.apache.cassandra.replication.MutationJournal; import org.apache.cassandra.replication.MutationTrackingService; import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.schema.MemtableParams; import org.apache.cassandra.schema.ReplicationType; import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.schema.TableParams; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.concurrent.OpOrder; @@ -106,6 +108,8 @@ public class SplitDomainMemtableTest TableMetadata metadata = TableMetadata.builder(ks, "tbl") .addPartitionKeyColumn("k", Int32Type.instance) .addRegularColumn("v", Int32Type.instance) + // test relies on Memtable#lastToken, so we explicitly set skiplist + .params(TableParams.builder().memtable(MemtableParams.get("skiplist")).build()) .build(); SchemaLoader.createKeyspace(ks, KeyspaceParams.simple(1, ReplicationType.tracked), metadata); ColumnFamilyStore cfs = Keyspace.open(ks).getColumnFamilyStore("tbl"); @@ -228,6 +232,7 @@ public class SplitDomainMemtableTest TableMetadata metadata = TableMetadata.builder(ks, "tbl") .addPartitionKeyColumn("k", Int32Type.instance) .addRegularColumn("v", Int32Type.instance) + .params(TableParams.builder().memtable(MemtableParams.get("skiplist")).build()) .build(); SchemaLoader.createKeyspace(ks, KeyspaceParams.simple(1, ReplicationType.untracked), metadata); diff --git a/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java b/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java index 374afab289..c54e51efdb 100644 --- a/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java +++ b/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java @@ -21,6 +21,8 @@ import java.io.IOException; import java.util.ArrayList; import java.util.List; +import org.junit.AfterClass; +import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; @@ -71,6 +73,18 @@ public class CoordinatorLogTest MutationJournal.start(); } + @Before + public void beforeTest() + { + MutationJournal.instance().truncateForTesting(); + } + + @AfterClass + public static void tearDown() + { + MutationJournal.instance().truncateForTesting(); + } + private static Token tk(String key) { return new ByteOrderedPartitioner.BytesToken(ByteBufferUtil.bytes(key)); diff --git a/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java b/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java index d283eadede..97155161b2 100644 --- a/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java +++ b/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java @@ -22,6 +22,7 @@ import java.util.ArrayList; import java.util.List; import org.junit.Assert; +import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; @@ -41,9 +42,11 @@ import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.schema.SchemaTestUtil; import org.apache.cassandra.utils.ByteBufferUtil; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; public class UnreconciledMutationsTest { @@ -63,6 +66,13 @@ public class UnreconciledMutationsTest TABLE_ID = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; } + @AfterClass + public static void tearDown() + { + if (MutationJournal.instance() != null) + MutationJournal.instance().truncateForTesting(); + } + private static Token tokenFor(int key) { return new ByteOrderedPartitioner.BytesToken(ByteBufferUtil.bytes(key)); @@ -447,4 +457,86 @@ public class UnreconciledMutationsTest unreconciled.collect(mutation7.key().getToken(), TABLE_ID, false, loadedOffsets); assertEquals(List.of(6, 7), loadedOffsets.asList()); } + + @Test + public void testLoadFromJournalWithDroppedTable() + { + MutationJournal.start(); + MutationJournal.instance().truncateForTesting(); + + String dropKs = "drop_ks"; + String dropTbl = "drop_tbl"; + SchemaLoader.createKeyspace(dropKs, KeyspaceParams.simple(1), + TableMetadata.builder(dropKs, dropTbl) + .addPartitionKeyColumn("k", Int32Type.instance) + .addRegularColumn("v", Int32Type.instance) + .build()); + TableMetadata dropTableMetadata = Schema.instance.getTableMetadata(dropKs, dropTbl); + + CoordinatorLogId logId = new CoordinatorLogId(10, 1); + + Offsets.Mutable offsets1 = new Offsets.Mutable(logId); + offsets1.add(1, 2); + + Offsets.Mutable offsets2 = new Offsets.Mutable(logId); + + Node2OffsetsMap witnessed = new Node2OffsetsMap(); + witnessed.set(1, offsets1); + witnessed.set(2, offsets2); + + MutationId id1 = new MutationId(logId.asLong(), MutationId.sequenceId(1, 0)); + Mutation mutation1 = new RowUpdateBuilder(dropTableMetadata, 0, 1) + .add("v", 1) + .build() + .withMutationId(id1); + + TableMetadata tableMetadata = Schema.instance.getTableMetadata(KEYSPACE, TABLE); + MutationId id2 = new MutationId(logId.asLong(), MutationId.sequenceId(2, 0)); + Mutation mutation2 = new RowUpdateBuilder(tableMetadata, 0, 2) + .add("v", 2) + .build() + .withMutationId(id2); + + MutationJournal.instance().write(mutation1.id(), mutation1); + MutationJournal.instance().write(mutation2.id(), mutation2); + + SchemaTestUtil.announceTableDrop(dropKs, dropTbl); + + UnreconciledMutations unreconciled = UnreconciledMutations.loadFromJournal(witnessed, 1); + Offsets.Mutable loadedOffsets = new Offsets.Mutable(logId); + unreconciled.collect(mutation2.key().getToken(), TABLE_ID, false, loadedOffsets); + assertEquals(List.of(2, 2), loadedOffsets.asList()); + assertTrue(loadedOffsets.contains(2)); + assertEquals(1, loadedOffsets.offsetCount()); + } + + @Test + public void testReplayStaticSegmentsWithDroppedTable() + { + MutationJournal.start(); + MutationJournal.instance().truncateForTesting(); + + String dropKs = "replay_drop_ks"; + String dropTbl = "replay_drop_tbl"; + SchemaLoader.createKeyspace(dropKs, KeyspaceParams.simple(1), + TableMetadata.builder(dropKs, dropTbl) + .addPartitionKeyColumn("k", Int32Type.instance) + .addRegularColumn("v", Int32Type.instance) + .build()); + TableMetadata dropTableMetadata = Schema.instance.getTableMetadata(dropKs, dropTbl); + + CoordinatorLogId logId = new CoordinatorLogId(20, 1); + MutationId id1 = new MutationId(logId.asLong(), MutationId.sequenceId(1, 0)); + Mutation mutation1 = new RowUpdateBuilder(dropTableMetadata, 0, 1) + .add("v", 1) + .build() + .withMutationId(id1); + + MutationJournal.instance().write(mutation1.id(), mutation1); + MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty(); + + SchemaTestUtil.announceTableDrop(dropKs, dropTbl); + + MutationJournal.instance().replayStaticSegments(); + } } \ No newline at end of file --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
