fixing CASSANDRA-13123 test
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/8fad4cd5 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/8fad4cd5 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/8fad4cd5 Branch: refs/heads/trunk Commit: 8fad4cd59ea5ef6d111cfe67d9a4c0345c4d7fd7 Parents: 9e37967 Author: Blake Eggleston <bdeggles...@gmail.com> Authored: Thu Oct 26 11:05:02 2017 -0700 Committer: Blake Eggleston <bdeggles...@gmail.com> Committed: Thu Oct 26 13:33:35 2017 -0700 ---------------------------------------------------------------------- .../CommitLogSegmentBackpressureTest.java | 140 ++++++++++++++++ .../commitlog/CommitLogSegmentManagerTest.java | 167 ------------------- .../db/commitlog/CommitlogShutdownTest.java | 98 +++++++++++ 3 files changed, 238 insertions(+), 167 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/8fad4cd5/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java new file mode 100644 index 0000000..b651098 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java @@ -0,0 +1,140 @@ +/* + * + * 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.db.commitlog; + +import java.nio.ByteBuffer; +import java.util.Random; +import java.util.concurrent.Semaphore; + +import com.google.common.collect.ImmutableMap; + +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.Util; +import org.apache.cassandra.config.Config.CommitLogSync; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.config.ParameterizedClass; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.db.RowUpdateBuilder; +import org.apache.cassandra.db.compaction.CompactionManager; +import org.apache.cassandra.db.marshal.AsciiType; +import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.schema.KeyspaceParams; +import org.jboss.byteman.contrib.bmunit.BMRule; +import org.jboss.byteman.contrib.bmunit.BMRules; +import org.jboss.byteman.contrib.bmunit.BMUnitRunner; + +/** + * Since this test depends on byteman rules being setup during initialization, you shouldn't add tests to this class + */ +@RunWith(BMUnitRunner.class) +public class CommitLogSegmentBackpressureTest +{ + //Block commit log service from syncing + private static final Semaphore allowSync = new Semaphore(1); + + private static final String KEYSPACE1 = "CommitLogTest"; + private static final String STANDARD1 = "Standard1"; + private static final String STANDARD2 = "Standard2"; + + private final static byte[] entropy = new byte[1024 * 256]; + + @Test + @BMRules(rules = {@BMRule(name = "Acquire Semaphore before sync", + targetClass = "AbstractCommitLogService$1", + targetMethod = "run", + targetLocation = "AT INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync", + action = "org.apache.cassandra.db.commitlog.CommitLogSegmentBackpressureTest.allowSync.acquire()"), + @BMRule(name = "Release Semaphore after sync", + targetClass = "AbstractCommitLogService$1", + targetMethod = "run", + targetLocation = "AFTER INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync", + action = "org.apache.cassandra.db.commitlog.CommitLogSegmentBackpressureTest.allowSync.release()")}) + public void testCompressedCommitLogBackpressure() throws Throwable + { + // Perform all initialization before making CommitLog.Sync blocking + // Doing the initialization within the method guarantee that Byteman has performed its injections when we start + new Random().nextBytes(entropy); + DatabaseDescriptor.setCommitLogCompression(new ParameterizedClass("LZ4Compressor", ImmutableMap.of())); + DatabaseDescriptor.setCommitLogSegmentSize(1); + DatabaseDescriptor.setCommitLogSync(CommitLogSync.periodic); + DatabaseDescriptor.setCommitLogSyncPeriod(10 * 1000); + SchemaLoader.prepareServer(); + SchemaLoader.createKeyspace(KEYSPACE1, + KeyspaceParams.simple(1), + SchemaLoader.standardCFMD(KEYSPACE1, STANDARD1, 0, AsciiType.instance, BytesType.instance), + SchemaLoader.standardCFMD(KEYSPACE1, STANDARD2, 0, AsciiType.instance, BytesType.instance)); + + CompactionManager.instance.disableAutoCompaction(); + + ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1); + + final Mutation m = new RowUpdateBuilder(cfs1.metadata, 0, "k").clustering("bytes") + .add("val", ByteBuffer.wrap(entropy)) + .build(); + + Thread dummyThread = new Thread(() -> { + for (int i = 0; i < 20; i++) + CommitLog.instance.add(m); + }); + + try + { + // Makes sure any call to CommitLog.sync is blocking + allowSync.acquire(); + + dummyThread.start(); + + CommitLogSegmentManager clsm = CommitLog.instance.allocator; + + Util.spinAssertEquals(3, () -> clsm.getActiveSegments().size(), 5); + + Thread.sleep(1000); + + // Should only be able to create 3 segments (not 7) because it blocks waiting for truncation that never + // comes. + Assert.assertEquals(3, clsm.getActiveSegments().size()); + + clsm.getActiveSegments().forEach(segment -> clsm.recycleSegment(segment)); + + Util.spinAssertEquals(3, () -> clsm.getActiveSegments().size(), 5); + } + finally + { + // Allow the CommitLog.sync to perform normally. + allowSync.release(); + } + try + { + // Wait for the dummy thread to die + dummyThread.join(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/8fad4cd5/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerTest.java deleted file mode 100644 index 41f5ed5..0000000 --- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerTest.java +++ /dev/null @@ -1,167 +0,0 @@ -/* - * - * 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.db.commitlog; - -import java.io.File; -import java.nio.ByteBuffer; -import java.util.Random; -import java.util.UUID; -import java.util.concurrent.Semaphore; - -import com.google.common.collect.ImmutableMap; - -import org.junit.Assert; -import org.junit.Test; -import org.junit.runner.RunWith; - -import org.apache.cassandra.SchemaLoader; -import org.apache.cassandra.Util; -import org.apache.cassandra.config.Config.CommitLogSync; -import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.config.ParameterizedClass; -import org.apache.cassandra.db.ColumnFamilyStore; -import org.apache.cassandra.db.Keyspace; -import org.apache.cassandra.db.Mutation; -import org.apache.cassandra.db.RowUpdateBuilder; -import org.apache.cassandra.db.compaction.CompactionManager; -import org.apache.cassandra.db.marshal.AsciiType; -import org.apache.cassandra.db.marshal.BytesType; -import org.apache.cassandra.schema.KeyspaceParams; -import org.jboss.byteman.contrib.bmunit.BMRule; -import org.jboss.byteman.contrib.bmunit.BMRules; -import org.jboss.byteman.contrib.bmunit.BMUnitRunner; - -@RunWith(BMUnitRunner.class) -public class CommitLogSegmentManagerTest -{ - //Block commit log service from syncing - private static final Semaphore allowSync = new Semaphore(1); - - private static final String KEYSPACE1 = "CommitLogTest"; - private static final String STANDARD1 = "Standard1"; - private static final String STANDARD2 = "Standard2"; - - private final static byte[] entropy = new byte[1024 * 256]; - - @Test - @BMRules(rules = {@BMRule(name = "Acquire Semaphore before sync", - targetClass = "AbstractCommitLogService$1", - targetMethod = "run", - targetLocation = "AT INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync", - action = "org.apache.cassandra.db.commitlog.CommitLogSegmentManagerTest.allowSync.acquire()"), - @BMRule(name = "Release Semaphore after sync", - targetClass = "AbstractCommitLogService$1", - targetMethod = "run", - targetLocation = "AFTER INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync", - action = "org.apache.cassandra.db.commitlog.CommitLogSegmentManagerTest.allowSync.release()")}) - public void testCompressedCommitLogBackpressure() throws Throwable - { - // Perform all initialization before making CommitLog.Sync blocking - // Doing the initialization within the method guarantee that Byteman has performed its injections when we start - new Random().nextBytes(entropy); - DatabaseDescriptor.setCommitLogCompression(new ParameterizedClass("LZ4Compressor", ImmutableMap.of())); - DatabaseDescriptor.setCommitLogSegmentSize(1); - DatabaseDescriptor.setCommitLogSync(CommitLogSync.periodic); - DatabaseDescriptor.setCommitLogSyncPeriod(10 * 1000); - SchemaLoader.prepareServer(); - SchemaLoader.createKeyspace(KEYSPACE1, - KeyspaceParams.simple(1), - SchemaLoader.standardCFMD(KEYSPACE1, STANDARD1, 0, AsciiType.instance, BytesType.instance), - SchemaLoader.standardCFMD(KEYSPACE1, STANDARD2, 0, AsciiType.instance, BytesType.instance)); - - CompactionManager.instance.disableAutoCompaction(); - - ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1); - - final Mutation m = new RowUpdateBuilder(cfs1.metadata, 0, "k").clustering("bytes") - .add("val", ByteBuffer.wrap(entropy)) - .build(); - - Thread dummyThread = new Thread(() -> { - for (int i = 0; i < 20; i++) - CommitLog.instance.add(m); - }); - - try - { - // Makes sure any call to CommitLog.sync is blocking - allowSync.acquire(); - - dummyThread.start(); - - CommitLogSegmentManager clsm = CommitLog.instance.allocator; - - Util.spinAssertEquals(3, () -> clsm.getActiveSegments().size(), 5); - - Thread.sleep(1000); - - // Should only be able to create 3 segments (not 7) because it blocks waiting for truncation that never - // comes. - Assert.assertEquals(3, clsm.getActiveSegments().size()); - - clsm.getActiveSegments().forEach(segment -> clsm.recycleSegment(segment)); - - Util.spinAssertEquals(3, () -> clsm.getActiveSegments().size(), 5); - } - finally - { - // Allow the CommitLog.sync to perform normally. - allowSync.release(); - } - try - { - // Wait for the dummy thread to die - dummyThread.join(); - } - catch (InterruptedException e) - { - Thread.currentThread().interrupt(); - } - } - - @Test - @BMRule(name = "Make removing commitlog segments slow", - targetClass = "CommitLogSegment", - targetMethod = "discard", - action = "Thread.sleep(50)") - public void testShutdownWithPendingTasks() throws Throwable { - CommitLog.instance.resetUnsafe(true); - ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1); - - final Mutation m = new RowUpdateBuilder(cfs1.metadata, 0, "k") - .clustering("bytes") - .add("val", ByteBuffer.wrap(entropy)) - .build(); - - // force creating several commitlog files - for (int i = 0; i < 10; i++) { - CommitLog.instance.add(m); - } - - // schedule discarding completed segments and immediately issue a shutdown - UUID cfid = m.getColumnFamilyIds().iterator().next(); - CommitLog.instance.discardCompletedSegments(cfid, ReplayPosition.NONE, CommitLog.instance.getContext()); - CommitLog.instance.shutdownBlocking(); - - // the shutdown should block until all logs except the currently active one and perhaps a new, empty one are gone - Assert.assertTrue(new File(CommitLog.instance.location).listFiles().length <= 2); - } -} http://git-wip-us.apache.org/repos/asf/cassandra/blob/8fad4cd5/test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java new file mode 100644 index 0000000..ee3f111 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java @@ -0,0 +1,98 @@ +/* + * 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.db.commitlog; + +import java.io.File; +import java.nio.ByteBuffer; +import java.util.Random; +import java.util.UUID; + +import com.google.common.collect.ImmutableMap; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.config.ParameterizedClass; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.db.RowUpdateBuilder; +import org.apache.cassandra.db.compaction.CompactionManager; +import org.apache.cassandra.db.marshal.AsciiType; +import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.schema.KeyspaceParams; +import org.jboss.byteman.contrib.bmunit.BMRule; +import org.jboss.byteman.contrib.bmunit.BMUnitRunner; + +/** + * Since this test depends on byteman rules being setup during initialization, you shouldn't add tests to this class + */ +@RunWith(BMUnitRunner.class) +public class CommitlogShutdownTest +{ + private static final String KEYSPACE1 = "CommitLogTest"; + private static final String STANDARD1 = "Standard1"; + + private final static byte[] entropy = new byte[1024 * 256]; + + @Test + @BMRule(name = "Make removing commitlog segments slow", + targetClass = "CommitLogSegment", + targetMethod = "discard", + action = "Thread.sleep(50)") + public void testShutdownWithPendingTasks() throws Exception + { + new Random().nextBytes(entropy); + DatabaseDescriptor.setCommitLogCompression(new ParameterizedClass("LZ4Compressor", ImmutableMap.of())); + DatabaseDescriptor.setCommitLogSegmentSize(1); + DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.periodic); + DatabaseDescriptor.setCommitLogSyncPeriod(10 * 1000); + SchemaLoader.prepareServer(); + SchemaLoader.createKeyspace(KEYSPACE1, + KeyspaceParams.simple(1), + SchemaLoader.standardCFMD(KEYSPACE1, STANDARD1, 0, AsciiType.instance, BytesType.instance)); + + CompactionManager.instance.disableAutoCompaction(); + + CommitLog.instance.resetUnsafe(true); + ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1); + + final Mutation m = new RowUpdateBuilder(cfs1.metadata, 0, "k") + .clustering("bytes") + .add("val", ByteBuffer.wrap(entropy)) + .build(); + + // force creating several commitlog files + for (int i = 0; i < 10; i++) + { + CommitLog.instance.add(m); + } + + // schedule discarding completed segments and immediately issue a shutdown + UUID cfid = m.getColumnFamilyIds().iterator().next(); + CommitLog.instance.discardCompletedSegments(cfid, ReplayPosition.NONE, CommitLog.instance.getContext()); + CommitLog.instance.shutdownBlocking(); + + // the shutdown should block until all logs except the currently active one and perhaps a new, empty one are gone + Assert.assertTrue(new File(CommitLog.instance.location).listFiles().length <= 2); + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org For additional commands, e-mail: commits-h...@cassandra.apache.org
