Add GroupCommitLogService

patch by Yuji Ito and jasobrown; reviewed by Ariel Weisberg for CASSANDRA-13530
2017-11-13 07:59:10 -08:00 · 2017-11-13 07:59:10 -08:00 · f3f90c1896
parent 4c80eeece3
commit f3f90c1896
16 changed files with 697 additions and 114 deletions
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@ -0,0 +1,254 @@
+env_vars: &env_vars
+    JAVA_HOME: /usr/local/openjdk8u154-cassandra-b02
+    ANT_HOME: /usr/local/apache-ant-1.10.1
+    LANG: en_US.UTF-8
+    JDK_HOME: /usr/local/openjdk8u154-cassandra-b02
+    JAVA8_HOME: /usr/local/openjdk8u154-cassandra-b02
+    JAVA7_HOME: /usr/local/openjdk7u82-cassandra-b02
+    DEBUG: true
+    PRINT_DEBUG: true
+    KEEP_TEST_DIR: true
+    CASSANDRA_DIR: /home/cassandra/cassandra
+    DEFAULT_DIR: /home/cassandra/cassandra-dtest
+    PYTHONIOENCODING: utf-8
+    PYTHONUNBUFFERED: true
+    CASS_DRIVER_NO_EXTENSIONS: true
+    CASS_DRIVER_NO_CYTHON: true
+    CCM_MAX_HEAP_SIZE: 3072M
+    CCM_HEAP_NEWSIZE: 256M
+version: 2
+jobs:
+  build:
+    resource_class: xlarge
+    working_directory: ~/
+    shell: /bin/bash -eo pipefail -l
+    docker:
+      - image: kjellman/cassandra-test:0.1.2
+        environment:
+            <<: *env_vars
+    steps:
+      - run:
+          name: Log Environment Information
+          command: |
+              echo '*** id ***'
+              id
+              echo '*** cat /proc/cpuinfo ***'
+              cat /proc/cpuinfo
+              echo '*** free -m ***'
+              free -m
+              echo '*** df -m ***'
+              df -m
+              echo '*** ifconfig -a ***'
+              ifconfig -a
+              echo '*** uname -a ***'
+              uname -a
+              echo '*** mount ***'
+              mount
+              echo '*** env ***'
+              env
+      - run:
+          name: Clone Cassandra Repository (via git)
+          command: |
+            export LANG=en_US.UTF-8
+            git clone --single-branch --depth 1 --branch $CIRCLE_BRANCH git://github.com/$CIRCLE_PROJECT_USERNAME/$CIRCLE_PROJECT_REPONAME.git ~/cassandra
+      - run:
+          name: Build Cassandra
+          command: |
+            export LANG=en_US.UTF-8
+            export JAVA_TOOL_OPTIONS="-Dfile.encoding=UTF8"
+            export PATH=$PATH:$ANT_HOME/bin:$JAVA_HOME/bin
+            cd ~/cassandra
+            # Loop to prevent failure due to maven-ant-tasks not downloading a jar..
+            for x in $(seq 1 3); do
+                ${ANT_HOME}/bin/ant clean jar
+                RETURN="$?"
+                if [ "${RETURN}" -eq "0" ]; then
+                    break
+                fi
+            done
+            # Exit, if we didn't build successfully
+            if [ "${RETURN}" -ne "0" ]; then
+                echo "Build failed with exit code: ${RETURN}"
+                exit ${RETURN}
+            fi
+          no_output_timeout: 20m
+      - persist_to_workspace:
+            root: /home/cassandra
+            paths:
+                - cassandra
+                - .m2
+  unit_tests:
+    resource_class: xlarge
+    parallelism: 100
+    working_directory: ~/
+    shell: /bin/bash -eo pipefail -l
+    docker:
+      - image: kjellman/cassandra-test:0.1.2
+        environment:
+            <<: *env_vars
+    steps:
+      - attach_workspace:
+          at: /home/cassandra
+      - run:
+          name: Determine Tests to Run
+          no_output_timeout: 10m
+          command: |
+            # reminder: this code (along with all the steps) is independently executed on every circle container
+            # so the goal here is to get the circleci script to return the tests *this* container will run
+            # which we do via the `circleci` cli tool.
+ 
+            export LANG=en_US.UTF-8
+            rm -fr ~/cassandra-dtest/upgrade_tests
+            echo "***java tests***"
+
+            # get all of our unit test filenames
+            set -eo pipefail && circleci tests glob "$HOME/cassandra/test/unit/**/*.java" > /tmp/all_java_unit_tests.txt
+
+            # split up the unit tests into groups based on the number of containers we have
+            set -eo pipefail && circleci tests split --split-by=timings --timings-type=filename --index=${CIRCLE_NODE_INDEX} --total=${CIRCLE_NODE_TOTAL} /tmp/all_java_unit_tests.txt > /tmp/java_tests_${CIRCLE_NODE_INDEX}.txt
+            set -eo pipefail && cat /tmp/java_tests_${CIRCLE_NODE_INDEX}.txt | cut -c 37-1000000 | grep "Test\.java$" > /tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt
+            echo "** /tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt"
+            cat /tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt
+      - run:
+         name: Run Unit Tests
+         command: |
+            export LANG=en_US.UTF-8
+            export JAVA_TOOL_OPTIONS="-Dfile.encoding=UTF8"
+            export PATH=$PATH:$ANT_HOME/bin:$JAVA_HOME/bin
+
+            cd ~/cassandra
+            ant testclasslist -Dtest.classlistfile=/tmp/java_tests_${CIRCLE_NODE_INDEX}_final.txt || true
+         no_output_timeout: 15m
+      - store_test_results:
+            path: ~/cassandra/build/test/output/
+      - store_artifacts:
+          path: ~/cassandra/build/test/output
+          destination: junitxml
+      - store_artifacts:
+          path: ~/cassandra/build/test/logs
+          destination: logs
+  dtests:
+    resource_class: xlarge
+    parallelism: 100
+    working_directory: ~/
+    shell: /bin/bash -eo pipefail -l
+    docker:
+      - image: kjellman/cassandra-test:0.1.2
+        environment:
+            <<: *env_vars
+    steps:
+      - attach_workspace:
+          at: /home/cassandra
+      - run:
+          name: Clone Cassandra dtest Repository (via git)
+          command: |
+            export LANG=en_US.UTF-8
+            git clone --single-branch --depth 1 git://github.com/jasobrown/cassandra-dtest.git ~/cassandra-dtest
+      - run:
+          name: Configure virtualenv and python Dependencies
+          command: |
+            # note, this should be super quick as all dependencies should be pre-installed in the docker image
+            # if additional dependencies were added to requirmeents.txt and the docker image hasn't been updated
+            # we'd have to install it here at runtime -- which will make things slow, so do yourself a favor and
+            # rebuild the docker image! (it automatically pulls the latest requirements.txt on build)
+            export LANG=en_US.UTF-8
+            source ~/env/bin/activate
+            export PATH=$PATH:$ANT_HOME/bin:$JAVA_HOME/bin
+            export CASS_DRIVER_NO_EXTENSIONS=true
+            export CASS_DRIVER_NO_CYTHON=true
+            pip install -r ~/cassandra-dtest/requirements.txt
+            pip freeze
+      - run:
+          name: Determine Tests to Run
+          no_output_timeout: 10m
+          command: |
+            # reminder: this code (along with all the steps) is independently executed on every circle container
+            # so the goal here is to get the circleci script to return the tests *this* container will run
+            # which we do via the `circleci` cli tool.
+ 
+            export LANG=en_US.UTF-8
+            echo "***python tests***"
+            cd cassandra-dtest
+            export CASSANDRA_DIR=/home/cassandra/cassandra
+            source ~/env/bin/activate
+            export PATH=$PATH:$ANT_HOME/bin:$JAVA_HOME/bin
+
+            echo "***regular python tests***"
+            set -eo pipefail && ./run_dtests.py --vnodes true --nose-options='--verbosity=3 --dtest-collect-only' 2>&1 | tee /tmp/pythontestlist
+            set -eo pipefail && cat /tmp/pythontestlist | grep -v 'upgrade_tests' | grep "... ok" | rev | cut -c 8- | rev > /tmp/testlist_trimmed
+            set -eo pipefail && circleci tests split --split-by=timings --timings-type=classname /tmp/testlist_trimmed > /tmp/python_tests.txt
+            cat /tmp/python_tests.txt
+            cat /tmp/python_tests.txt | grep -v "\-novnode" | grep -v "\-offheap" > /tmp/regular_python_tests.txt
+            cat /tmp/regular_python_tests.txt | tr '\n' ' ' > /tmp/regular_python_tests_final.txt
+            cat /tmp/regular_python_tests_final.txt
+
+            echo "***novnode python tests***"
+            rm /tmp/pythontestlist || true
+            ./run_dtests.py --vnodes false --nose-options='--verbosity=3 --dtest-collect-only --attr=!resource-intensive' 2>&1 | tee -a /tmp/pythontestlist
+            set -eo pipefail && cat /tmp/pythontestlist | grep -v 'upgrade_tests' | grep "... ok" | rev | cut -c 8- | rev > /tmp/testlist_trimmed
+            set -eo pipefail && circleci tests split --split-by=timings --timings-type=classname /tmp/testlist_trimmed > /tmp/novnode_python_tests.txt
+            cat /tmp/novnode_python_tests.txt | tr '\n' ' ' > /tmp/novnode_python_tests_final.txt
+            cat /tmp/novnode_python_tests_final.txt
+
+            # echo "***offheap python tests***"
+            # rm /tmp/pythontestlist || true
+            # TEST_TAG=offheap ./run_dtests.py --vnodes true --nose-options='--verbosity=3 --dtest-collect-only --attr=!resource-intensive' 2>&1 | grep -v "upgrade_tests." | tee -a /tmp/pythontestlist
+            # cat /tmp/python_tests.txt | grep "\-offheap" | rev | cut -c 9- | rev > /tmp/offheap_python_tests.txt
+            # cat /tmp/offheap_python_tests.txt | tr '\n' ' ' > /tmp/offheap_python_tests_final.txt
+            # cat /tmp/offheap_python_tests_final.txt
+      - run:
+          name: Run dtests
+          no_output_timeout: 30m
+          command: |
+            echo "cat /tmp/regular_python_tests_final.txt"
+            cat /tmp/regular_python_tests_final.txt
+
+            source ~/env/bin/activate
+            export PATH=$PATH:$ANT_HOME/bin:$JAVA_HOME/bin
+
+            cd ~/cassandra-dtest
+            mkdir -p /tmp/dtest
+            export NUM_TOKENS="32"
+
+            echo "env: $(env)"
+            echo "** done env"
+            ./run_dtests.py --vnodes true --nose-options="--verbosity=3 --with-xunit --nocapture" `cat /tmp/regular_python_tests_final.txt` 2>&1 | tee /tmp/dtest/stdout.txt || true
+
+            # store the test results somewhere where we can pick them up
+            mv nosetests.xml nosetests-regular.xml
+            mkdir -p /tmp/results/ || true
+            find ~/cassandra-dtest/ -name "*.xml" -exec cp {} /tmp/results/ \; 
+      - run:
+          name: Run dtests (without vnodes)
+          no_output_timeout: 30m
+          command: |
+            source ~/env/bin/activate
+            export PATH=$PATH:$ANT_HOME/bin:$JAVA_HOME/bin
+
+            cd ~/cassandra-dtest
+            export TEST_TAG="novnodes"
+            ./run_dtests.py --vnodes false --nose-options="--verbosity=3 --with-dtestxunit --nocapture --attr=!resource-intensive" `cat /tmp/novnode_python_tests_final.txt` 2>&1 | tee /tmp/dtest/stdout-novnodes.txt || true
+
+            # store the test results somewhere where we can pick them up
+            mv nosetests.xml nosetests-novnode.xml
+            mkdir -p /tmp/results/ || true
+            find ~/cassandra-dtest/ -name "*.xml" -exec cp {} /tmp/results/ \; 
+      - store_test_results:
+          path: /tmp/results/
+      - store_artifacts:
+          path: /tmp/dtest
+          destination: dtest
+      - store_artifacts:
+          path: ~/cassandra-dtest/logs
+          destination: dtest_logs
+workflows:
+    version: 2
+    build_and_run_tests:
+        jobs:
+            - build
+            - unit_tests:
+                  requires:
+                      - build
+            - dtests:
+                  requires:
+                      - unit_tests
--- a/NEWS.txt
+++ b/NEWS.txt
@ -82,6 +82,11 @@ Upgrading
      the per token range view building status is stored in the new table
      `system.view_builds_in_progress`. The old table `system.views_builds_in_progress`
      is no longer used and can be removed. See CASSANDRA-12245 for more details.
+	- Config option commitlog_sync_batch_window_in_ms has been deprecated as it's
+	  documentation has been incorrect and the setting itself near useless.
+	  Batch mode remains a valid commit log mode, however.
+	- There is a new commit log mode, group, which is similar to batch mode
+	  but blocks for up to a configurable number of milliseconds between disk flushes.

 Materialized Views
 -------------------
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@ -358,19 +358,22 @@ counter_cache_save_period: 7200
 # If not set, the default directory is $CASSANDRA_HOME/data/saved_caches.
 # saved_caches_directory: /var/lib/cassandra/saved_caches

-# commitlog_sync may be either "periodic" or "batch." 
+# commitlog_sync may be either "periodic", "group", or "batch." 
 # 
 # When in batch mode, Cassandra won't ack writes until the commit log
-# has been fsynced to disk.  It will wait
-# commitlog_sync_batch_window_in_ms milliseconds between fsyncs.
-# This window should be kept short because the writer threads will
-# be unable to do extra work while waiting.  (You may need to increase
-# concurrent_writes for the same reason.)
+# has been flushed to disk.  Each incoming write will trigger the flush task.
+# commitlog_sync_batch_window_in_ms is a deprecated value. Previously it had
+# almost no value, and is being removed.
 #
-# commitlog_sync: batch
 # commitlog_sync_batch_window_in_ms: 2
 #
-# the other option is "periodic" where writes may be acked immediately
+# group mode is similar to batch mode, where Cassandra will not ack writes
+# until the commit log has been flushed to disk. The difference is group
+# mode will wait up to commitlog_sync_group_window_in_ms between flushes.
+#
+# commitlog_sync_group_window_in_ms: 1000
+#
+# the default option is "periodic" where writes may be acked immediately
 # and the CommitLog is simply synced every commitlog_sync_period_in_ms
 # milliseconds. 
 commitlog_sync: periodic
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@ -185,7 +185,12 @@ public class Config
    public String commitlog_directory;
    public Integer commitlog_total_space_in_mb;
    public CommitLogSync commitlog_sync;
+
+    /**
+     * @deprecated since 4.0 This value was near useless, and we're not using it anymore
+     */
    public double commitlog_sync_batch_window_in_ms = Double.NaN;
+    public double commitlog_sync_group_window_in_ms = Double.NaN;
    public int commitlog_sync_period_in_ms;
    public int commitlog_segment_size_in_mb = 32;
    public ParameterizedClass commitlog_compression;
@ -383,7 +388,8 @@ public class Config
    public enum CommitLogSync
    {
        periodic,
-        batch
+        batch,
+        group
    }
    public enum InternodeCompression
    {
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@ -332,15 +332,23 @@ public class DatabaseDescriptor

        if (conf.commitlog_sync == Config.CommitLogSync.batch)
        {
-            if (Double.isNaN(conf.commitlog_sync_batch_window_in_ms) || conf.commitlog_sync_batch_window_in_ms <= 0d)
-            {
-                throw new ConfigurationException("Missing value for commitlog_sync_batch_window_in_ms: positive double value expected.", false);
-            }
-            else if (conf.commitlog_sync_period_in_ms != 0)
+            if (conf.commitlog_sync_period_in_ms != 0)
            {
                throw new ConfigurationException("Batch sync specified, but commitlog_sync_period_in_ms found. Only specify commitlog_sync_batch_window_in_ms when using batch sync", false);
            }
-            logger.debug("Syncing log with a batch window of {}", conf.commitlog_sync_batch_window_in_ms);
+            logger.debug("Syncing log with batch mode");
+        }
+        else if (conf.commitlog_sync == CommitLogSync.group)
+        {
+            if (Double.isNaN(conf.commitlog_sync_group_window_in_ms) || conf.commitlog_sync_group_window_in_ms <= 0d)
+            {
+                throw new ConfigurationException("Missing value for commitlog_sync_group_window_in_ms: positive double value expected.", false);
+            }
+            else if (conf.commitlog_sync_period_in_ms != 0)
+            {
+                throw new ConfigurationException("Group sync specified, but commitlog_sync_period_in_ms found. Only specify commitlog_sync_group_window_in_ms when using group sync", false);
+            }
+            logger.debug("Syncing log with a group window of {}", conf.commitlog_sync_period_in_ms);
        }
        else
        {
@ -1784,14 +1792,14 @@ public class DatabaseDescriptor
        conf.native_transport_max_concurrent_connections_per_ip = native_transport_max_concurrent_connections_per_ip;
    }

-    public static double getCommitLogSyncBatchWindow()
+    public static double getCommitLogSyncGroupWindow()
    {
-        return conf.commitlog_sync_batch_window_in_ms;
+        return conf.commitlog_sync_group_window_in_ms;
    }

-    public static void setCommitLogSyncBatchWindow(double windowMillis)
+    public static void setCommitLogSyncGroupWindow(double windowMillis)
    {
-        conf.commitlog_sync_batch_window_in_ms = windowMillis;
+        conf.commitlog_sync_group_window_in_ms = windowMillis;
    }

    public static int getCommitLogSyncPeriod()
--- a/src/java/org/apache/cassandra/db/commitlog/BatchCommitLogService.java
+++ b/src/java/org/apache/cassandra/db/commitlog/BatchCommitLogService.java
@ -17,13 +17,18 @@
 */
 package org.apache.cassandra.db.commitlog;

-import org.apache.cassandra.config.DatabaseDescriptor;
-
 class BatchCommitLogService extends AbstractCommitLogService
 {
+    /**
+     * Batch mode does not rely on the sync thread in {@link AbstractCommitLogService} to wake up for triggering
+     * the disk sync. Instead we trigger it explicitly in {@link #maybeWaitForSync(CommitLogSegment.Allocation)}.
+     * This value here is largely irrelevant, but should high enough so the sync thread is not continually waking up.
+     */
+    private static final int POLL_TIME_MILLIS = 1000;
+
    public BatchCommitLogService(CommitLog commitLog)
    {
-        super(commitLog, "COMMIT-LOG-WRITER", (int) DatabaseDescriptor.getCommitLogSyncBatchWindow());
+        super(commitLog, "COMMIT-LOG-WRITER", POLL_TIME_MILLIS);
    }

    protected void maybeWaitForSync(CommitLogSegment.Allocation alloc)
--- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
+++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
@ -104,9 +104,20 @@ public class CommitLog implements CommitLogMBean
        this.archiver = archiver;
        metrics = new CommitLogMetrics();

-        executor = DatabaseDescriptor.getCommitLogSync() == Config.CommitLogSync.batch
-                ? new BatchCommitLogService(this)
-                : new PeriodicCommitLogService(this);
+        switch (DatabaseDescriptor.getCommitLogSync())
+        {
+            case periodic:
+                executor = new PeriodicCommitLogService(this);
+                break;
+            case batch:
+                executor = new BatchCommitLogService(this);
+                break;
+            case group:
+                executor = new GroupCommitLogService(this);
+                break;
+            default:
+                throw new IllegalArgumentException("Unknown commitlog service type: " + DatabaseDescriptor.getCommitLogSync());
+        }

        segmentManager = DatabaseDescriptor.isCDCEnabled()
                         ? new CommitLogSegmentManagerCDC(this, DatabaseDescriptor.getCommitLogLocation())
--- a/src/java/org/apache/cassandra/db/commitlog/GroupCommitLogService.java
+++ b/src/java/org/apache/cassandra/db/commitlog/GroupCommitLogService.java
@ -0,0 +1,43 @@
+/*
+ * 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 org.apache.cassandra.config.DatabaseDescriptor;
+
+/**
+ * A commitlog service that will block returning an ACK back to the a coordinator/client
+ * for a minimum amount of time as we wait until the the commit log segment is flushed.
+ */
+public class GroupCommitLogService extends AbstractCommitLogService
+{
+    public GroupCommitLogService(CommitLog commitLog)
+    {
+        super(commitLog, "GROUP-COMMIT-LOG-WRITER", (int) DatabaseDescriptor.getCommitLogSyncGroupWindow());
+    }
+
+    protected void maybeWaitForSync(CommitLogSegment.Allocation alloc)
+    {
+        // wait until record has been safely persisted to disk
+        pending.incrementAndGet();
+        // wait for commitlog_sync_group_window_in_ms
+        alloc.awaitDiskSync(commitLog.metrics.waitingOnCommit);
+        pending.decrementAndGet();
+    }
+}
+
--- a/test/long/org/apache/cassandra/db/commitlog/BatchCommitLogStressTest.java
+++ b/test/long/org/apache/cassandra/db/commitlog/BatchCommitLogStressTest.java
@ -0,0 +1,37 @@
+/*
+ * 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 org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.security.EncryptionContext;
+
+@RunWith(Parameterized.class)
+public class BatchCommitLogStressTest extends CommitLogStressTest
+{
+    public BatchCommitLogStressTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
+    {
+        super(commitLogCompression, encryptionContext);
+        DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.batch);
+    }
+}
--- a/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java
+++ b/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java
@ -33,7 +33,11 @@ import com.google.common.util.concurrent.RateLimiter;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.BeforeClass;
+import org.junit.Ignore;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;

 import io.netty.util.concurrent.FastThreadLocalThread;
 import org.apache.cassandra.SchemaLoader;
@ -45,6 +49,9 @@ import org.apache.cassandra.db.Mutation;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.db.partitions.PartitionUpdate;
 import org.apache.cassandra.db.rows.*;
+import org.apache.cassandra.io.compress.DeflateCompressor;
+import org.apache.cassandra.io.compress.LZ4Compressor;
+import org.apache.cassandra.io.compress.SnappyCompressor;
 import org.apache.cassandra.io.util.DataInputBuffer;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.schema.Schema;
@ -52,7 +59,8 @@ import org.apache.cassandra.security.EncryptionContext;
 import org.apache.cassandra.security.EncryptionContextGenerator;


-public class CommitLogStressTest
+@Ignore
+public abstract class CommitLogStressTest
 {
    static
    {
@ -80,55 +88,68 @@ public class CommitLogStressTest
        return hash;
    }

-    public static void main(String[] args) throws Exception
+    // TODO:JEB resolve this
+//    public static void main(String[] args) throws Exception
+//    {
+//        try
+//        {
+//            if (args.length >= 1)
+//            {
+//                NUM_THREADS = Integer.parseInt(args[0]);
+//                System.out.println("Setting num threads to: " + NUM_THREADS);
+//            }
+//
+//            if (args.length >= 2)
+//            {
+//                numCells = Integer.parseInt(args[1]);
+//                System.out.println("Setting num cells to: " + numCells);
+//            }
+//
+//            if (args.length >= 3)
+//            {
+//                cellSize = Integer.parseInt(args[1]);
+//                System.out.println("Setting cell size to: " + cellSize + " be aware the source corpus may be small");
+//            }
+//
+//            if (args.length >= 4)
+//            {
+//                rateLimit = Integer.parseInt(args[1]);
+//                System.out.println("Setting per thread rate limit to: " + rateLimit);
+//            }
+//            initialize();
+//
+//            for (Object[] params : buildParameterizedVariants())
+//            {
+//                ParameterizedClass commitLogCompression = (ParameterizedClass) params[0];
+//                EncryptionContext encryptionContext = (EncryptionContext) params[1];
+//                CommitLogStressTest tester = new CommitLogStressTest(commitLogCompression, encryptionContext);
+//                tester.cleanDir();
+//                tester.testFixedSize();
+//            }
+//        }
+//        catch (Throwable e)
+//        {
+//            e.printStackTrace(System.err);
+//        }
+//        finally
+//        {
+//            System.exit(0);
+//        }
+//    }
+
+    private boolean failed = false;
+    private volatile boolean stop = false;
+    private boolean randomSize = false;
+    private boolean discardedRun = false;
+    private CommitLogPosition discardedPos;
+
+    public CommitLogStressTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
    {
-        try
-        {
-            if (args.length >= 1)
-            {
-                NUM_THREADS = Integer.parseInt(args[0]);
-                System.out.println("Setting num threads to: " + NUM_THREADS);
-            }
-
-            if (args.length >= 2)
-            {
-                numCells = Integer.parseInt(args[1]);
-                System.out.println("Setting num cells to: " + numCells);
-            }
-
-            if (args.length >= 3)
-            {
-                cellSize = Integer.parseInt(args[1]);
-                System.out.println("Setting cell size to: " + cellSize + " be aware the source corpus may be small");
-            }
-
-            if (args.length >= 4)
-            {
-                rateLimit = Integer.parseInt(args[1]);
-                System.out.println("Setting per thread rate limit to: " + rateLimit);
-            }
-            initialize();
-
-            CommitLogStressTest tester = new CommitLogStressTest();
-            tester.cleanDir();
-            tester.testFixedSize();
-        }
-        catch (Throwable e)
-        {
-            e.printStackTrace(System.err);
-        }
-        finally
-        {
-            System.exit(0);
-        }
+        DatabaseDescriptor.setCommitLogCompression(commitLogCompression);
+        DatabaseDescriptor.setEncryptionContext(encryptionContext);
+        DatabaseDescriptor.setCommitLogSegmentSize(32);
    }

-    boolean failed = false;
-    volatile boolean stop = false;
-    boolean randomSize = false;
-    boolean discardedRun = false;
-    CommitLogPosition discardedPos;
-
    @BeforeClass
    static public void initialize() throws IOException
    {
@ -166,12 +187,23 @@ public class CommitLogStressTest
        }
    }

+    @Parameters()
+    public static Collection<Object[]> buildParameterizedVariants()
+    {
+        return Arrays.asList(new Object[][]{
+        {null, EncryptionContextGenerator.createDisabledContext()}, // No compression, no encryption
+        {null, EncryptionContextGenerator.createContext(true)}, // Encryption
+        { new ParameterizedClass(LZ4Compressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
+        { new ParameterizedClass(SnappyCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
+        { new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}});
+    }
+
    @Test
    public void testRandomSize() throws Exception
    {
        randomSize = true;
        discardedRun = false;
-        testAllLogConfigs();
+        testLog();
    }

    @Test
@ -179,8 +211,7 @@ public class CommitLogStressTest
    {
        randomSize = false;
        discardedRun = false;
-
-        testAllLogConfigs();
+        testLog();
    }

    @Test
@ -188,45 +219,18 @@ public class CommitLogStressTest
    {
        randomSize = true;
        discardedRun = true;
-
-        testAllLogConfigs();
+        testLog();
    }

-    public void testAllLogConfigs() throws IOException, InterruptedException
+    private void testLog() throws IOException, InterruptedException
    {
-        failed = false;
-        DatabaseDescriptor.setCommitLogSyncBatchWindow(1);
-        DatabaseDescriptor.setCommitLogSyncPeriod(30);
-        DatabaseDescriptor.setCommitLogSegmentSize(32);
-
-        // test plain vanilla commit logs (the choice of 98% of users)
-        testLog(null, EncryptionContextGenerator.createDisabledContext());
-
-        // test the compression types
-        testLog(new ParameterizedClass("LZ4Compressor", null), EncryptionContextGenerator.createDisabledContext());
-        testLog(new ParameterizedClass("SnappyCompressor", null), EncryptionContextGenerator.createDisabledContext());
-        testLog(new ParameterizedClass("DeflateCompressor", null), EncryptionContextGenerator.createDisabledContext());
-
-        // test the encrypted commit log
-        testLog(null, EncryptionContextGenerator.createContext(true));
-    }
-
-    public void testLog(ParameterizedClass compression, EncryptionContext encryptionContext) throws IOException, InterruptedException
-    {
-        DatabaseDescriptor.setCommitLogCompression(compression);
-        DatabaseDescriptor.setEncryptionContext(encryptionContext);
-
        String originalDir = DatabaseDescriptor.getCommitLogLocation();
        try
        {
            DatabaseDescriptor.setCommitLogLocation(location);
-            for (CommitLogSync sync : CommitLogSync.values())
-            {
-                DatabaseDescriptor.setCommitLogSync(sync);
-                CommitLog commitLog = new CommitLog(CommitLogArchiver.disabled()).start();
-                testLog(commitLog);
-                assert !failed;
-            }
+            CommitLog commitLog = new CommitLog(CommitLogArchiver.disabled()).start();
+            testLog(commitLog);
+            assert !failed;
        }
        finally
        {
@ -234,7 +238,7 @@ public class CommitLogStressTest
        }
    }

-    public void testLog(CommitLog commitLog) throws IOException, InterruptedException {
+    private void testLog(CommitLog commitLog) throws IOException, InterruptedException {
        System.out.format("\nTesting commit log size %.0fmb, compressor: %s, encryption enabled: %b, sync %s%s%s\n",
                           mb(DatabaseDescriptor.getCommitLogSegmentSize()),
                           commitLog.configuration.getCompressorName(),
@ -346,7 +350,7 @@ public class CommitLogStressTest
        Assert.assertTrue(ratios.isEmpty());
    }

-    public ScheduledExecutorService startThreads(final CommitLog commitLog, final List<CommitlogThread> threads)
+    private ScheduledExecutorService startThreads(final CommitLog commitLog, final List<CommitlogThread> threads)
    {
        stop = false;
        for (int ii = 0; ii < NUM_THREADS; ii++) {
@ -405,7 +409,7 @@ public class CommitLogStressTest
        return maxMemory / (1024 * 1024);
    }

-    public static ByteBuffer randomBytes(int quantity, Random tlr)
+    private static ByteBuffer randomBytes(int quantity, Random tlr)
    {
        ByteBuffer slice = ByteBuffer.allocate(quantity);
        ByteBuffer source = dataSource.duplicate();
@ -428,7 +432,7 @@ public class CommitLogStressTest

        volatile CommitLogPosition clsp;

-        public CommitlogThread(CommitLog commitLog, Random rand)
+        CommitlogThread(CommitLog commitLog, Random rand)
        {
            this.commitLog = commitLog;
            this.random = rand;
@ -525,7 +529,7 @@ public class CommitLogStressTest
        }
    }

-    class DummyHandler implements CommitLogReadHandler
+    static class DummyHandler implements CommitLogReadHandler
    {
        public boolean shouldSkipSegmentOnError(CommitLogReadException exception) throws IOException { return false; }

--- a/test/long/org/apache/cassandra/db/commitlog/GroupCommitLogStressTest.java
+++ b/test/long/org/apache/cassandra/db/commitlog/GroupCommitLogStressTest.java
@ -0,0 +1,38 @@
+/*
+ * 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 org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.security.EncryptionContext;
+
+@RunWith(Parameterized.class)
+public class GroupCommitLogStressTest extends CommitLogStressTest
+{
+    public GroupCommitLogStressTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
+    {
+        super(commitLogCompression, encryptionContext);
+        DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.group);
+        DatabaseDescriptor.setCommitLogSyncGroupWindow(1);
+    }
+}
--- a/test/long/org/apache/cassandra/db/commitlog/PeriodicCommitLogStressTest.java
+++ b/test/long/org/apache/cassandra/db/commitlog/PeriodicCommitLogStressTest.java
@ -0,0 +1,39 @@
+/*
+ * 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 org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.security.EncryptionContext;
+
+@RunWith(Parameterized.class)
+public class PeriodicCommitLogStressTest extends CommitLogStressTest
+{
+    public PeriodicCommitLogStressTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
+    {
+        super(commitLogCompression, encryptionContext);
+        DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.periodic);
+        DatabaseDescriptor.setCommitLogSyncPeriod(30);
+    }
+
+}
--- a/test/unit/org/apache/cassandra/db/commitlog/BatchCommitLogTest.java
+++ b/test/unit/org/apache/cassandra/db/commitlog/BatchCommitLogTest.java
@ -0,0 +1,42 @@
+/*
+ * 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 org.junit.BeforeClass;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.security.EncryptionContext;
+
+public class BatchCommitLogTest extends CommitLogTest
+{
+    public BatchCommitLogTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
+    {
+        super(commitLogCompression, encryptionContext);
+    }
+
+    @BeforeClass
+    public static void setCommitLogModeDetails()
+    {
+        DatabaseDescriptor.daemonInitialization();
+        DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.batch);
+        beforeClass();
+    }
+}
--- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTest.java
+++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTest.java
@ -71,8 +71,9 @@ import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;

+@Ignore
@RunWith(Parameterized.class)
-public class CommitLogTest
+public abstract class CommitLogTest
 {
    private static final String KEYSPACE1 = "CommitLogTest";
    private static final String KEYSPACE2 = "CommitLogTestNonDurable";
@ -99,7 +100,6 @@ public class CommitLogTest
            {new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}});
    }

-    @BeforeClass
    public static void beforeClass() throws ConfigurationException
    {
        // Disable durable writes for system keyspaces to prevent system mutations, e.g. sstable_activity,
--- a/test/unit/org/apache/cassandra/db/commitlog/GroupCommitLogTest.java
+++ b/test/unit/org/apache/cassandra/db/commitlog/GroupCommitLogTest.java
@ -0,0 +1,43 @@
+/*
+ * 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 org.junit.BeforeClass;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.security.EncryptionContext;
+
+public class GroupCommitLogTest extends CommitLogTest
+{
+    public GroupCommitLogTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
+    {
+        super(commitLogCompression, encryptionContext);
+    }
+
+    @BeforeClass
+    public static void setCommitLogModeDetails()
+    {
+        DatabaseDescriptor.daemonInitialization();
+        DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.group);
+        DatabaseDescriptor.setCommitLogSyncGroupWindow(1);
+        beforeClass();
+    }
+}
--- a/test/unit/org/apache/cassandra/db/commitlog/PeriodicCommitLogTest.java
+++ b/test/unit/org/apache/cassandra/db/commitlog/PeriodicCommitLogTest.java
@ -0,0 +1,45 @@
+/*
+ * 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 org.junit.BeforeClass;
+import org.junit.Ignore;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.security.EncryptionContext;
+
+@Ignore
+public class PeriodicCommitLogTest extends CommitLogTest
+{
+    public PeriodicCommitLogTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
+    {
+        super(commitLogCompression, encryptionContext);
+    }
+
+    @BeforeClass
+    public static void setCommitLogModeDetails()
+    {
+        DatabaseDescriptor.daemonInitialization();
+        DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.periodic);
+        DatabaseDescriptor.setCommitLogSyncPeriod(10 * 1000);
+        beforeClass();
+    }
+}