| /* |
| * 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.cassandra.batchlog; |
| |
| import java.io.IOException; |
| import java.net.InetAddress; |
| import java.util.*; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.TimeUnit; |
| |
| import com.google.common.collect.Lists; |
| |
| import org.junit.*; |
| |
| import org.apache.cassandra.SchemaLoader; |
| import org.apache.cassandra.Util; |
| import org.apache.cassandra.Util.PartitionerSwitcher; |
| import org.apache.cassandra.config.CFMetaData; |
| import org.apache.cassandra.config.DatabaseDescriptor; |
| import org.apache.cassandra.config.Schema; |
| import org.apache.cassandra.config.SchemaConstants; |
| import org.apache.cassandra.cql3.UntypedResultSet; |
| import org.apache.cassandra.db.ColumnFamilyStore; |
| import org.apache.cassandra.db.DecoratedKey; |
| import org.apache.cassandra.db.Keyspace; |
| import org.apache.cassandra.db.Mutation; |
| import org.apache.cassandra.db.RowUpdateBuilder; |
| import org.apache.cassandra.db.SystemKeyspace; |
| import org.apache.cassandra.db.commitlog.CommitLogPosition; |
| import org.apache.cassandra.db.marshal.BytesType; |
| import org.apache.cassandra.db.partitions.ImmutableBTreePartition; |
| import org.apache.cassandra.db.partitions.PartitionUpdate; |
| import org.apache.cassandra.db.rows.Row; |
| import org.apache.cassandra.dht.Murmur3Partitioner; |
| import org.apache.cassandra.exceptions.ConfigurationException; |
| import org.apache.cassandra.locator.TokenMetadata; |
| import org.apache.cassandra.net.MessagingService; |
| import org.apache.cassandra.service.StorageService; |
| import org.apache.cassandra.schema.KeyspaceParams; |
| import org.apache.cassandra.utils.ByteBufferUtil; |
| import org.apache.cassandra.utils.FBUtilities; |
| import org.apache.cassandra.utils.UUIDGen; |
| |
| import static org.apache.cassandra.cql3.QueryProcessor.executeInternal; |
| import static org.junit.Assert.*; |
| |
| public class BatchlogManagerTest |
| { |
| private static final String KEYSPACE1 = "BatchlogManagerTest1"; |
| private static final String CF_STANDARD1 = "Standard1"; |
| private static final String CF_STANDARD2 = "Standard2"; |
| private static final String CF_STANDARD3 = "Standard3"; |
| private static final String CF_STANDARD4 = "Standard4"; |
| private static final String CF_STANDARD5 = "Standard5"; |
| |
| static PartitionerSwitcher sw; |
| |
| @BeforeClass |
| public static void defineSchema() throws ConfigurationException |
| { |
| DatabaseDescriptor.daemonInitialization(); |
| sw = Util.switchPartitioner(Murmur3Partitioner.instance); |
| SchemaLoader.prepareServer(); |
| SchemaLoader.createKeyspace(KEYSPACE1, |
| KeyspaceParams.simple(1), |
| SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD1, 1, BytesType.instance), |
| SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD2, 1, BytesType.instance), |
| SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD3, 1, BytesType.instance), |
| SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD4, 1, BytesType.instance), |
| SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD5, 1, BytesType.instance)); |
| } |
| |
| @AfterClass |
| public static void cleanup() |
| { |
| sw.close(); |
| } |
| |
| @Before |
| @SuppressWarnings("deprecation") |
| public void setUp() throws Exception |
| { |
| TokenMetadata metadata = StorageService.instance.getTokenMetadata(); |
| InetAddress localhost = InetAddress.getByName("127.0.0.1"); |
| metadata.updateNormalToken(Util.token("A"), localhost); |
| metadata.updateHostId(UUIDGen.getTimeUUID(), localhost); |
| Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.BATCHES).truncateBlocking(); |
| Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.LEGACY_BATCHLOG).truncateBlocking(); |
| } |
| |
| @Test |
| public void testDelete() |
| { |
| ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1); |
| CFMetaData cfm = cfs.metadata; |
| new RowUpdateBuilder(cfm, FBUtilities.timestampMicros(), ByteBufferUtil.bytes("1234")) |
| .clustering("c") |
| .add("val", "val" + 1234) |
| .build() |
| .applyUnsafe(); |
| |
| DecoratedKey dk = cfs.decorateKey(ByteBufferUtil.bytes("1234")); |
| ImmutableBTreePartition results = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, dk).build()); |
| Iterator<Row> iter = results.iterator(); |
| assert iter.hasNext(); |
| |
| Mutation mutation = new Mutation(PartitionUpdate.fullPartitionDelete(cfm, |
| dk, |
| FBUtilities.timestampMicros(), |
| FBUtilities.nowInSeconds())); |
| mutation.applyUnsafe(); |
| |
| Util.assertEmpty(Util.cmd(cfs, dk).build()); |
| } |
| |
| @Test |
| public void testReplay() throws Exception |
| { |
| testReplay(false); |
| } |
| |
| @Test |
| public void testLegacyReplay() throws Exception |
| { |
| testReplay(true); |
| } |
| |
| @SuppressWarnings("deprecation") |
| private static void testReplay(boolean legacy) throws Exception |
| { |
| long initialAllBatches = BatchlogManager.instance.countAllBatches(); |
| long initialReplayedBatches = BatchlogManager.instance.getTotalBatchesReplayed(); |
| |
| CFMetaData cfm = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1).metadata; |
| |
| // Generate 1000 mutations (100 batches of 10 mutations each) and put them all into the batchlog. |
| // Half batches (50) ready to be replayed, half not. |
| for (int i = 0; i < 100; i++) |
| { |
| List<Mutation> mutations = new ArrayList<>(10); |
| for (int j = 0; j < 10; j++) |
| { |
| mutations.add(new RowUpdateBuilder(cfm, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(i)) |
| .clustering("name" + j) |
| .add("val", "val" + j) |
| .build()); |
| } |
| |
| long timestamp = i < 50 |
| ? (System.currentTimeMillis() - BatchlogManager.getBatchlogTimeout()) |
| : (System.currentTimeMillis() + BatchlogManager.getBatchlogTimeout()); |
| |
| if (legacy) |
| LegacyBatchlogMigrator.store(Batch.createLocal(UUIDGen.getTimeUUID(timestamp, i), timestamp * 1000, mutations), MessagingService.current_version); |
| else |
| BatchlogManager.store(Batch.createLocal(UUIDGen.getTimeUUID(timestamp, i), timestamp * 1000, mutations)); |
| } |
| |
| if (legacy) |
| { |
| Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.LEGACY_BATCHLOG).forceBlockingFlush(); |
| LegacyBatchlogMigrator.migrate(); |
| } |
| |
| // Flush the batchlog to disk (see CASSANDRA-6822). |
| Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.BATCHES).forceBlockingFlush(); |
| |
| assertEquals(100, BatchlogManager.instance.countAllBatches() - initialAllBatches); |
| assertEquals(0, BatchlogManager.instance.getTotalBatchesReplayed() - initialReplayedBatches); |
| |
| // Force batchlog replay and wait for it to complete. |
| BatchlogManager.instance.startBatchlogReplay().get(); |
| |
| // Ensure that the first half, and only the first half, got replayed. |
| assertEquals(50, BatchlogManager.instance.countAllBatches() - initialAllBatches); |
| assertEquals(50, BatchlogManager.instance.getTotalBatchesReplayed() - initialReplayedBatches); |
| |
| for (int i = 0; i < 100; i++) |
| { |
| String query = String.format("SELECT * FROM \"%s\".\"%s\" WHERE key = intAsBlob(%d)", KEYSPACE1, CF_STANDARD1, i); |
| UntypedResultSet result = executeInternal(query); |
| assertNotNull(result); |
| if (i < 50) |
| { |
| Iterator<UntypedResultSet.Row> it = result.iterator(); |
| assertNotNull(it); |
| for (int j = 0; j < 10; j++) |
| { |
| assertTrue(it.hasNext()); |
| UntypedResultSet.Row row = it.next(); |
| |
| assertEquals(ByteBufferUtil.bytes(i), row.getBytes("key")); |
| assertEquals("name" + j, row.getString("name")); |
| assertEquals("val" + j, row.getString("val")); |
| } |
| |
| assertFalse(it.hasNext()); |
| } |
| else |
| { |
| assertTrue(result.isEmpty()); |
| } |
| } |
| |
| // Ensure that no stray mutations got somehow applied. |
| UntypedResultSet result = executeInternal(String.format("SELECT count(*) FROM \"%s\".\"%s\"", KEYSPACE1, CF_STANDARD1)); |
| assertNotNull(result); |
| assertEquals(500, result.one().getLong("count")); |
| } |
| |
| @Test |
| public void testTruncatedReplay() throws InterruptedException, ExecutionException |
| { |
| CFMetaData cf2 = Schema.instance.getCFMetaData(KEYSPACE1, CF_STANDARD2); |
| CFMetaData cf3 = Schema.instance.getCFMetaData(KEYSPACE1, CF_STANDARD3); |
| // Generate 2000 mutations (1000 batchlog entries) and put them all into the batchlog. |
| // Each batchlog entry with a mutation for Standard2 and Standard3. |
| // In the middle of the process, 'truncate' Standard2. |
| for (int i = 0; i < 1000; i++) |
| { |
| Mutation mutation1 = new RowUpdateBuilder(cf2, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(i)) |
| .clustering("name" + i) |
| .add("val", "val" + i) |
| .build(); |
| Mutation mutation2 = new RowUpdateBuilder(cf3, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(i)) |
| .clustering("name" + i) |
| .add("val", "val" + i) |
| .build(); |
| |
| List<Mutation> mutations = Lists.newArrayList(mutation1, mutation2); |
| |
| // Make sure it's ready to be replayed, so adjust the timestamp. |
| long timestamp = System.currentTimeMillis() - BatchlogManager.getBatchlogTimeout(); |
| |
| if (i == 500) |
| SystemKeyspace.saveTruncationRecord(Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD2), |
| timestamp, |
| CommitLogPosition.NONE); |
| |
| // Adjust the timestamp (slightly) to make the test deterministic. |
| if (i >= 500) |
| timestamp++; |
| else |
| timestamp--; |
| |
| BatchlogManager.store(Batch.createLocal(UUIDGen.getTimeUUID(timestamp, i), FBUtilities.timestampMicros(), mutations)); |
| } |
| |
| // Flush the batchlog to disk (see CASSANDRA-6822). |
| Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.BATCHES).forceBlockingFlush(); |
| |
| // Force batchlog replay and wait for it to complete. |
| BatchlogManager.instance.startBatchlogReplay().get(); |
| |
| // We should see half of Standard2-targeted mutations written after the replay and all of Standard3 mutations applied. |
| for (int i = 0; i < 1000; i++) |
| { |
| UntypedResultSet result = executeInternal(String.format("SELECT * FROM \"%s\".\"%s\" WHERE key = intAsBlob(%d)", KEYSPACE1, CF_STANDARD2,i)); |
| assertNotNull(result); |
| if (i >= 500) |
| { |
| assertEquals(ByteBufferUtil.bytes(i), result.one().getBytes("key")); |
| assertEquals("name" + i, result.one().getString("name")); |
| assertEquals("val" + i, result.one().getString("val")); |
| } |
| else |
| { |
| assertTrue(result.isEmpty()); |
| } |
| } |
| |
| for (int i = 0; i < 1000; i++) |
| { |
| UntypedResultSet result = executeInternal(String.format("SELECT * FROM \"%s\".\"%s\" WHERE key = intAsBlob(%d)", KEYSPACE1, CF_STANDARD3, i)); |
| assertNotNull(result); |
| assertEquals(ByteBufferUtil.bytes(i), result.one().getBytes("key")); |
| assertEquals("name" + i, result.one().getString("name")); |
| assertEquals("val" + i, result.one().getString("val")); |
| } |
| } |
| |
| @Test |
| @SuppressWarnings("deprecation") |
| public void testConversion() throws Exception |
| { |
| long initialAllBatches = BatchlogManager.instance.countAllBatches(); |
| long initialReplayedBatches = BatchlogManager.instance.getTotalBatchesReplayed(); |
| CFMetaData cfm = Schema.instance.getCFMetaData(KEYSPACE1, CF_STANDARD4); |
| |
| // Generate 1400 version 2.0 mutations and put them all into the batchlog. |
| // Half ready to be replayed, half not. |
| for (int i = 0; i < 1400; i++) |
| { |
| Mutation mutation = new RowUpdateBuilder(cfm, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(i)) |
| .clustering("name" + i) |
| .add("val", "val" + i) |
| .build(); |
| |
| long timestamp = i < 700 |
| ? (System.currentTimeMillis() - BatchlogManager.getBatchlogTimeout()) |
| : (System.currentTimeMillis() + BatchlogManager.getBatchlogTimeout()); |
| |
| |
| Mutation batchMutation = LegacyBatchlogMigrator.getStoreMutation(Batch.createLocal(UUIDGen.getTimeUUID(timestamp, i), |
| TimeUnit.MILLISECONDS.toMicros(timestamp), |
| Collections.singleton(mutation)), |
| MessagingService.VERSION_20); |
| assertTrue(LegacyBatchlogMigrator.isLegacyBatchlogMutation(batchMutation)); |
| LegacyBatchlogMigrator.handleLegacyMutation(batchMutation); |
| } |
| |
| // Mix in 100 current version mutations, 50 ready for replay. |
| for (int i = 1400; i < 1500; i++) |
| { |
| Mutation mutation = new RowUpdateBuilder(cfm, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(i)) |
| .clustering("name" + i) |
| .add("val", "val" + i) |
| .build(); |
| |
| long timestamp = i < 1450 |
| ? (System.currentTimeMillis() - BatchlogManager.getBatchlogTimeout()) |
| : (System.currentTimeMillis() + BatchlogManager.getBatchlogTimeout()); |
| |
| |
| BatchlogManager.store(Batch.createLocal(UUIDGen.getTimeUUID(timestamp, i), |
| FBUtilities.timestampMicros(), |
| Collections.singleton(mutation))); |
| } |
| |
| // Flush the batchlog to disk (see CASSANDRA-6822). |
| Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.BATCHES).forceBlockingFlush(); |
| |
| assertEquals(1500, BatchlogManager.instance.countAllBatches() - initialAllBatches); |
| assertEquals(0, BatchlogManager.instance.getTotalBatchesReplayed() - initialReplayedBatches); |
| |
| UntypedResultSet result = executeInternal(String.format("SELECT count(*) FROM \"%s\".\"%s\"", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.LEGACY_BATCHLOG)); |
| assertNotNull(result); |
| assertEquals("Count in blog legacy", 0, result.one().getLong("count")); |
| result = executeInternal(String.format("SELECT count(*) FROM \"%s\".\"%s\"", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.BATCHES)); |
| assertNotNull(result); |
| assertEquals("Count in blog", 1500, result.one().getLong("count")); |
| |
| // Force batchlog replay and wait for it to complete. |
| BatchlogManager.instance.performInitialReplay(); |
| |
| // Ensure that the first half, and only the first half, got replayed. |
| assertEquals(750, BatchlogManager.instance.countAllBatches() - initialAllBatches); |
| assertEquals(750, BatchlogManager.instance.getTotalBatchesReplayed() - initialReplayedBatches); |
| |
| for (int i = 0; i < 1500; i++) |
| { |
| result = executeInternal(String.format("SELECT * FROM \"%s\".\"%s\" WHERE key = intAsBlob(%d)", KEYSPACE1, CF_STANDARD4, i)); |
| assertNotNull(result); |
| if (i < 700 || i >= 1400 && i < 1450) |
| { |
| assertEquals(ByteBufferUtil.bytes(i), result.one().getBytes("key")); |
| assertEquals("name" + i, result.one().getString("name")); |
| assertEquals("val" + i, result.one().getString("val")); |
| } |
| else |
| { |
| assertTrue("Present at " + i, result.isEmpty()); |
| } |
| } |
| |
| // Ensure that no stray mutations got somehow applied. |
| result = executeInternal(String.format("SELECT count(*) FROM \"%s\".\"%s\"", KEYSPACE1, CF_STANDARD4)); |
| assertNotNull(result); |
| assertEquals(750, result.one().getLong("count")); |
| |
| // Ensure batchlog is left as expected. |
| result = executeInternal(String.format("SELECT count(*) FROM \"%s\".\"%s\"", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.BATCHES)); |
| assertNotNull(result); |
| assertEquals("Count in blog after initial replay", 750, result.one().getLong("count")); |
| result = executeInternal(String.format("SELECT count(*) FROM \"%s\".\"%s\"", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.LEGACY_BATCHLOG)); |
| assertNotNull(result); |
| assertEquals("Count in blog legacy after initial replay ", 0, result.one().getLong("count")); |
| } |
| |
| @Test |
| public void testAddBatch() throws IOException |
| { |
| long initialAllBatches = BatchlogManager.instance.countAllBatches(); |
| CFMetaData cfm = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD5).metadata; |
| |
| long timestamp = (System.currentTimeMillis() - DatabaseDescriptor.getWriteRpcTimeout() * 2) * 1000; |
| UUID uuid = UUIDGen.getTimeUUID(); |
| |
| // Add a batch with 10 mutations |
| List<Mutation> mutations = new ArrayList<>(10); |
| for (int j = 0; j < 10; j++) |
| { |
| mutations.add(new RowUpdateBuilder(cfm, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(j)) |
| .clustering("name" + j) |
| .add("val", "val" + j) |
| .build()); |
| } |
| |
| |
| BatchlogManager.store(Batch.createLocal(uuid, timestamp, mutations)); |
| Assert.assertEquals(initialAllBatches + 1, BatchlogManager.instance.countAllBatches()); |
| |
| String query = String.format("SELECT count(*) FROM %s.%s where id = %s", |
| SchemaConstants.SYSTEM_KEYSPACE_NAME, |
| SystemKeyspace.BATCHES, |
| uuid); |
| UntypedResultSet result = executeInternal(query); |
| assertNotNull(result); |
| assertEquals(1L, result.one().getLong("count")); |
| } |
| |
| @Test |
| public void testRemoveBatch() |
| { |
| long initialAllBatches = BatchlogManager.instance.countAllBatches(); |
| CFMetaData cfm = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD5).metadata; |
| |
| long timestamp = (System.currentTimeMillis() - DatabaseDescriptor.getWriteRpcTimeout() * 2) * 1000; |
| UUID uuid = UUIDGen.getTimeUUID(); |
| |
| // Add a batch with 10 mutations |
| List<Mutation> mutations = new ArrayList<>(10); |
| for (int j = 0; j < 10; j++) |
| { |
| mutations.add(new RowUpdateBuilder(cfm, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(j)) |
| .clustering("name" + j) |
| .add("val", "val" + j) |
| .build()); |
| } |
| |
| // Store the batch |
| BatchlogManager.store(Batch.createLocal(uuid, timestamp, mutations)); |
| Assert.assertEquals(initialAllBatches + 1, BatchlogManager.instance.countAllBatches()); |
| |
| // Remove the batch |
| BatchlogManager.remove(uuid); |
| |
| assertEquals(initialAllBatches, BatchlogManager.instance.countAllBatches()); |
| |
| String query = String.format("SELECT count(*) FROM %s.%s where id = %s", |
| SchemaConstants.SYSTEM_KEYSPACE_NAME, |
| SystemKeyspace.BATCHES, |
| uuid); |
| UntypedResultSet result = executeInternal(query); |
| assertNotNull(result); |
| assertEquals(0L, result.one().getLong("count")); |
| } |
| |
| // CASSANRDA-9223 |
| @Test |
| public void testReplayWithNoPeers() throws Exception |
| { |
| StorageService.instance.getTokenMetadata().removeEndpoint(InetAddress.getByName("127.0.0.1")); |
| |
| long initialAllBatches = BatchlogManager.instance.countAllBatches(); |
| long initialReplayedBatches = BatchlogManager.instance.getTotalBatchesReplayed(); |
| |
| CFMetaData cfm = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1).metadata; |
| |
| long timestamp = (System.currentTimeMillis() - DatabaseDescriptor.getWriteRpcTimeout() * 2) * 1000; |
| UUID uuid = UUIDGen.getTimeUUID(); |
| |
| // Add a batch with 10 mutations |
| List<Mutation> mutations = new ArrayList<>(10); |
| for (int j = 0; j < 10; j++) |
| { |
| mutations.add(new RowUpdateBuilder(cfm, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(j)) |
| .clustering("name" + j) |
| .add("val", "val" + j) |
| .build()); |
| } |
| BatchlogManager.store(Batch.createLocal(uuid, timestamp, mutations)); |
| assertEquals(1, BatchlogManager.instance.countAllBatches() - initialAllBatches); |
| |
| // Flush the batchlog to disk (see CASSANDRA-6822). |
| Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.BATCHES).forceBlockingFlush(); |
| |
| assertEquals(1, BatchlogManager.instance.countAllBatches() - initialAllBatches); |
| assertEquals(0, BatchlogManager.instance.getTotalBatchesReplayed() - initialReplayedBatches); |
| |
| // Force batchlog replay and wait for it to complete. |
| BatchlogManager.instance.startBatchlogReplay().get(); |
| |
| // Replay should be cancelled as there are no peers in the ring. |
| assertEquals(1, BatchlogManager.instance.countAllBatches() - initialAllBatches); |
| } |
| } |
