Support a means of logging all queries as they were invoked.

Patch by Ariel Weisberg; Reviewed by Blake Eggleston

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0
+ * Support a means of logging all queries as they were invoked (CASSANDRA-13983)
  * Presize collections (CASSANDRA-13760)
  * Add GroupCommitLogService (CASSANDRA-13530)
  * Parallelize initial materialized view build (CASSANDRA-12245)

NEWS.txt

@@ -34,6 +34,14 @@ New features
      threads is specified by the property `cassandra.yaml:concurrent_materialized_view_builders`.
      This property can be modified at runtime through both JMX and the new `setconcurrentviewbuilders`
      and `getconcurrentviewbuilders` nodetool commands. See CASSANDRA-12245 for more details.
+   - There is now a binary full query log based on Chronicle Queue that can be controlled using
+     nodetool enablefullquerylog, disablefullquerylog, and resetfullquerylog. The log 
+     contains all queries invoked, approximate time they were invoked, any parameters necessary
+     to bind wildcard values, and all query options. A human readable version of the log can be
+     dumped or tailed using the new bin/fqltool utility. The full query log is designed to be safe
+     to use in production and limits utilization of heap memory and disk space with limits
+     you can specify when enabling the log.
+     See nodetool and fqltool help text for more information.
 
 Upgrading
 ---------

bin/fqltool

@@ -0,0 +1,88 @@
+#!/bin/sh
+
+# 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.
+
+if [ "x$CASSANDRA_INCLUDE" = "x" ]; then
+    # Locations (in order) to use when searching for an include file.
+    for include in "`dirname "$0"`/cassandra.in.sh" \
+                   "$HOME/.cassandra.in.sh" \
+                   /usr/share/cassandra/cassandra.in.sh \
+                   /usr/local/share/cassandra/cassandra.in.sh \
+                   /opt/cassandra/cassandra.in.sh; do
+        if [ -r "$include" ]; then
+            . "$include"
+            break
+        fi
+    done
+elif [ -r "$CASSANDRA_INCLUDE" ]; then
+    . "$CASSANDRA_INCLUDE"
+fi
+
+# Use JAVA_HOME if set, otherwise look for java in PATH
+if [ -x "$JAVA_HOME/bin/java" ]; then
+    JAVA="$JAVA_HOME/bin/java"
+else
+    JAVA="`which java`"
+fi
+
+if [ "x$JAVA" = "x" ]; then
+    echo "Java executable not found (hint: set JAVA_HOME)" >&2
+    exit 1
+fi
+
+if [ -z "$CASSANDRA_CONF" -o -z "$CLASSPATH" ]; then
+    echo "You must set the CASSANDRA_CONF and CLASSPATH vars" >&2
+    exit 1
+fi
+
+# Run cassandra-env.sh to pick up JMX_PORT
+if [ -f "$CASSANDRA_CONF/cassandra-env.sh" ]; then
+    JVM_OPTS_SAVE=$JVM_OPTS
+    MAX_HEAP_SIZE_SAVE=$MAX_HEAP_SIZE
+    . "$CASSANDRA_CONF/cassandra-env.sh"
+    MAX_HEAP_SIZE=$MAX_HEAP_SIZE_SAVE
+    JVM_OPTS=$JVM_OPTS_SAVE
+fi
+
+# JMX Port passed via cmd line args (-p 9999 / --port 9999 / --port=9999)
+# should override the value from cassandra-env.sh
+ARGS=""
+JVM_ARGS=""
+while true
+do
+  if [ ! $1 ]; then break; fi
+  case $1 in
+    -D*)
+      JVM_ARGS="$JVM_ARGS $1"
+      ;;
+    *)
+      ARGS="$ARGS $1"
+      ;;
+  esac
+  shift
+done
+
+if [ "x$MAX_HEAP_SIZE" = "x" ]; then
+    MAX_HEAP_SIZE="512m"
+fi
+
+"$JAVA" $JAVA_AGENT -ea -da:net.openhft... -cp "$CLASSPATH" $JVM_OPTS -Xmx$MAX_HEAP_SIZE \
+        -Dlog4j.configurationFile=log4j2-tools.xml \
+        $JVM_ARGS \
+        org.apache.cassandra.tools.FullQueryLogTool $ARGS
+
+# vi:ai sw=4 ts=4 tw=0 et

bin/fqltool.bat

@@ -0,0 +1,36 @@
+@REM
+@REM Licensed to the Apache Software Foundation (ASF) under one or more
+@REM contributor license agreements. See the NOTICE file distributed with
+@REM this work for additional information regarding copyright ownership.
+@REM The ASF licenses this file to You under the Apache License, Version 2.0
+@REM (the "License"); you may not use this file except in compliance with
+@REM the License. You may obtain a copy of the License at
+@REM
+@REM http://www.apache.org/licenses/LICENSE-2.0
+@REM
+@REM Unless required by applicable law or agreed to in writing, software
+@REM distributed under the License is distributed on an "AS IS" BASIS,
+@REM WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+@REM See the License for the specific language governing permissions and
+@REM limitations under the License.
+
+@echo off
+if "%OS%" == "Windows_NT" setlocal
+
+pushd "%~dp0"
+call cassandra.in.bat
+
+if NOT DEFINED JAVA_HOME goto :err
+
+set CASSANDRA_PARAMS=%CASSANDRA_PARAMS% -Dcassandra.logdir="%CASSANDRA_HOME%\logs"
+
+"%JAVA_HOME%\bin\java" -cp %CASSANDRA_CLASSPATH% %CASSANDRA_PARAMS% -Dlog4j.configurationFile=log4j2-tools.xml org.apache.cassandra.tools.FullQueryLogTool %*
+goto finally
+
+:err
+echo The JAVA_HOME environment variable must be set to run this program!
+pause
+
+:finally
+ENDLOCAL & set RC=%ERRORLEVEL%
+exit /B %RC%

build.xml

@@ -114,6 +114,13 @@
 
     <property name="ecj.version" value="4.4.2"/>
 
+    <!-- https://mvnrepository.com/artifact/net.openhft/chronicle-bom/1.15.6 -->
+    <property name="chronicle-queue.version" value="4.6.55" />
+    <property name="chronicle-core.version" value="1.9.21" />
+    <property name="chronicle-bytes.version" value="1.10.1" />
+    <property name="chronicle-wire.version" value="1.10.1" />
+    <property name="chronicle-threads.version" value="1.9.1" />
+
     <condition property="maven-ant-tasks.jar.exists">
       <available file="${build.dir}/maven-ant-tasks-${maven-ant-tasks.version}.jar" />
     </condition>
@@ -422,6 +429,11 @@
           <dependency groupId="org.mindrot" artifactId="jbcrypt" version="0.3m" />
           <dependency groupId="io.airlift" artifactId="airline" version="0.8" />
           <dependency groupId="io.netty" artifactId="netty-all" version="4.1.14.Final" />
+          <dependency groupId="net.openhft" artifactId="chronicle-queue" version="${chronicle-queue.version}"/>
+          <dependency groupId="net.openhft" artifactId="chronicle-core" version="${chronicle-core.version}"/>
+          <dependency groupId="net.openhft" artifactId="chronicle-bytes" version="${chronicle-bytes.version}"/>
+          <dependency groupId="net.openhft" artifactId="chronicle-wire" version="${chronicle-wire.version}"/>
+          <dependency groupId="net.openhft" artifactId="chronicle-threads" version="${chronicle-threads.version}"/>
           <dependency groupId="com.google.code.findbugs" artifactId="jsr305" version="2.0.2" />
           <dependency groupId="com.clearspring.analytics" artifactId="stream" version="2.5.2" />
 	  <!-- UPDATE AND UNCOMMENT ON THE DRIVER RELEASE, BEFORE 4.0 RELEASE
@@ -620,6 +632,11 @@
         <dependency groupId="com.github.jbellis" artifactId="jamm"/>
 
         <dependency groupId="io.netty" artifactId="netty-all"/>
+        <dependency groupId="net.openhft" artifactId="chronicle-queue" version="${chronicle-queue.version}"/>
+        <dependency groupId="net.openhft" artifactId="chronicle-core" version="${chronicle-core.version}"/>
+        <dependency groupId="net.openhft" artifactId="chronicle-bytes" version="${chronicle-bytes.version}"/>
+        <dependency groupId="net.openhft" artifactId="chronicle-wire" version="${chronicle-wire.version}"/>
+        <dependency groupId="net.openhft" artifactId="chronicle-threads" version="${chronicle-threads.version}"/>
         <dependency groupId="joda-time" artifactId="joda-time"/>
         <dependency groupId="org.fusesource" artifactId="sigar"/>
         <dependency groupId="org.eclipse.jdt.core.compiler" artifactId="ecj"/>

conf/cassandra.yaml

@@ -1051,9 +1051,9 @@ transparent_data_encryption_options:
     key_alias: testing:1
     # CBC IV length for AES needs to be 16 bytes (which is also the default size)
     # iv_length: 16
-    key_provider: 
+    key_provider:
       - class_name: org.apache.cassandra.security.JKSKeyProvider
-        parameters: 
+        parameters:
           - keystore: conf/.keystore
             keystore_password: cassandra
             store_type: JCEKS
@@ -1165,3 +1165,8 @@ back_pressure_strategy:
 # level for writes without timing out. This is different from the consistency level requested by
 # each write which may be lower in order to facilitate availability.
 # ideal_consistency_level: EACH_QUORUM
+
+# Path to write full query log data to when the full query log is enabled
+# The full query log will recrusively delete the contents of this path at
+# times. Don't place links in this directory to other parts of the filesystem.
+#full_query_log_dir: /tmp/cassandrafullquerylog

conf/jvm.options

@@ -86,6 +86,10 @@
 # enable assertions. highly suggested for correct application functionality.
 -ea
 
+# disable assertions for net.openhft.** because it runs out of memory by design
+# if enabled and run for more than just brief testing
+-da:net.openhft...
+
 # enable thread priorities, primarily so we can give periodic tasks
 # a lower priority to avoid interfering with client workload
 -XX:+UseThreadPriorities

lib/licenses/chronicle-bytes-1.10.1.txt

@@ -0,0 +1,14 @@
+
+== Copyright 2016 higherfrequencytrading.com
+
+Licensed 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.

lib/licenses/chronicle-core-1.9.21.txt

@@ -0,0 +1,14 @@
+
+== Copyright 2016 higherfrequencytrading.com
+
+Licensed 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.

lib/licenses/chronicle-queue-4.6.55.txt

@@ -0,0 +1,14 @@
+
+== Copyright 2016 higherfrequencytrading.com
+
+Licensed 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.

lib/licenses/chronicle-threads-1.9.1.txt

@@ -0,0 +1,14 @@
+
+== Copyright 2016 higherfrequencytrading.com
+
+Licensed 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.

lib/licenses/chronicle-wire-1.10.1.txt

@@ -0,0 +1,14 @@
+
+== Copyright 2016 higherfrequencytrading.com
+
+Licensed 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.

src/java/org/apache/cassandra/config/Config.java

@@ -364,6 +364,8 @@ public class Config
     public RepairCommandPoolFullStrategy repair_command_pool_full_strategy = RepairCommandPoolFullStrategy.queue;
     public int repair_command_pool_size = concurrent_validations;
 
+    public String full_query_log_dir = null;
+
     /**
      * @deprecated migrate to {@link DatabaseDescriptor#isClientInitialized()}
      */

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -2474,4 +2474,9 @@ public class DatabaseDescriptor
     {
         return conf.repair_command_pool_full_strategy;
     }
+
+    public static String getFullQueryLogPath()
+    {
+        return  conf.full_query_log_dir;
+    }
 }

src/java/org/apache/cassandra/db/fullquerylog/FullQueryLogger.java

@@ -0,0 +1,530 @@
+/*
+ * 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.fullquerylog;
+
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Sets;
+import com.google.common.primitives.Ints;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.buffer.ByteBuf;
+import net.openhft.chronicle.bytes.BytesStore;
+import net.openhft.chronicle.queue.RollCycles;
+import net.openhft.chronicle.wire.ValueOut;
+import net.openhft.chronicle.wire.WireOut;
+import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.io.FSError;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.transport.CBUtil;
+import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.utils.NoSpamLogger;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Throwables;
+import org.apache.cassandra.utils.binlog.BinLog;
+import org.apache.cassandra.utils.concurrent.WeightedQueue;
+import org.github.jamm.MemoryLayoutSpecification;
+
+/**
+ * A logger that logs entire query contents after the query finishes (or times out).
+ */
+public class FullQueryLogger
+{
+    private static final int EMPTY_BYTEBUFFER_SIZE = Ints.checkedCast(ObjectSizes.sizeOnHeapExcludingData(ByteBuffer.allocate(0)));
+    private static final int EMPTY_LIST_SIZE = Ints.checkedCast(ObjectSizes.measureDeep(new ArrayList(0)));
+    private static final int EMPTY_BYTEBUF_SIZE;
+    private static final int OBJECT_HEADER_SIZE = MemoryLayoutSpecification.SPEC.getObjectHeaderSize();
+    static
+    {
+        int tempSize = 0;
+        ByteBuf buf = CBUtil.allocator.buffer(0, 0);
+        try
+        {
+            tempSize = Ints.checkedCast(ObjectSizes.measure(buf));
+        }
+        finally
+        {
+            buf.release();
+        }
+        EMPTY_BYTEBUF_SIZE = tempSize;
+    }
+
+    private static final Logger logger = LoggerFactory.getLogger(FullQueryLogger.class);
+    private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1, TimeUnit.MINUTES);
+    private static final NoSpamLogger.NoSpamLogStatement droppedSamplesStatement = noSpamLogger.getStatement("Dropped {} binary log samples", 1, TimeUnit.MINUTES);
+
+    public static final FullQueryLogger instance = new FullQueryLogger();
+
+    volatile BinLog binLog;
+    private volatile boolean blocking;
+    private Path path;
+
+    private final AtomicLong droppedSamplesSinceLastLog = new AtomicLong();
+
+    private FullQueryLogger()
+    {
+    }
+
+    /**
+     * Configure the global instance of the FullQueryLogger
+     * @param path Dedicated path where the FQL can store it's files.
+     * @param rollCycle How often to roll FQL log segments so they can potentially be reclaimed
+     * @param blocking Whether the FQL should block if the FQL falls behind or should drop log records
+     * @param maxQueueWeight Maximum weight of in memory queue for records waiting to be written to the file before blocking or dropping
+     * @param maxLogSize Maximum size of the rolled files to retain on disk before deleting the oldest file
+     */
+    public synchronized void configure(Path path, String rollCycle, boolean blocking, int maxQueueWeight, long maxLogSize)
+    {
+        Preconditions.checkNotNull(path, "path was null");
+        File pathAsFile = path.toFile();
+        Preconditions.checkNotNull(rollCycle, "rollCycle was null");
+        rollCycle = rollCycle.toUpperCase();
+
+        //Exists and is a directory or can be created
+        Preconditions.checkArgument((pathAsFile.exists() && pathAsFile.isDirectory()) || (!pathAsFile.exists() && pathAsFile.mkdirs()), "path exists and is not a directory or couldn't be created");
+        Preconditions.checkArgument(pathAsFile.canRead() && pathAsFile.canWrite() && pathAsFile.canExecute(), "path is not readable, writable, and executable");
+        Preconditions.checkNotNull(RollCycles.valueOf(rollCycle), "unrecognized roll cycle");
+        Preconditions.checkArgument(maxQueueWeight > 0, "maxQueueWeight must be > 0");
+        Preconditions.checkArgument(maxLogSize > 0, "maxLogSize must be > 0");
+        logger.info("Attempting to configure full query logger path: {} Roll cycle: {} Blocking: {} Max queue weight: {} Max log size:{}", path, rollCycle, blocking, maxQueueWeight, maxLogSize);
+        if (binLog != null)
+        {
+            logger.warn("Full query logger already configured. Ignoring requested configuration.");
+            throw new IllegalStateException("Already configured");
+        }
+
+        if (path.toFile().exists())
+        {
+            Throwable error = cleanDirectory(path.toFile(), null);
+            if (error != null)
+            {
+                throw new RuntimeException(error);
+            }
+        }
+
+        this.path = path;
+        this.blocking = blocking;
+        binLog = new BinLog(path, RollCycles.valueOf(rollCycle), maxQueueWeight, maxLogSize);
+        binLog.start();
+    }
+
+    /**
+     * Need the path as a parameter as well because if the process is restarted the config file might be the only
+     * location for retrieving the path to the full query log files, but JMX also allows you to specify a path
+     * that isn't persisted anywhere so we have to clean that one a well.
+     */
+    public synchronized void reset(String fullQueryLogPath)
+    {
+        try
+        {
+            Set<File> pathsToClean = Sets.newHashSet();
+
+            //First decide whether to clean the path configured in the YAML
+            if (fullQueryLogPath != null)
+            {
+                File fullQueryLogPathFile = new File(fullQueryLogPath);
+                if (fullQueryLogPathFile.exists())
+                {
+                    pathsToClean.add(fullQueryLogPathFile);
+                }
+            }
+
+            //Then decide whether to clean the last used path, possibly configured by JMX
+            if (path != null)
+            {
+                File pathFile = path.toFile();
+                if (pathFile.exists())
+                {
+                    pathsToClean.add(pathFile);
+                }
+            }
+
+            logger.info("Reset (and deactivation) of full query log requested.");
+            if (binLog != null)
+            {
+                logger.info("Stopping full query log. Cleaning {}.", pathsToClean);
+                binLog.stop();
+                binLog = null;
+            }
+            else
+            {
+                logger.info("Full query log already deactivated. Cleaning {}.", pathsToClean);
+            }
+
+            Throwable accumulate = null;
+            for (File f : pathsToClean)
+            {
+                accumulate = cleanDirectory(f, accumulate);
+            }
+            if (accumulate != null)
+            {
+                throw new RuntimeException(accumulate);
+            }
+        }
+        catch (Exception e)
+        {
+            if (e instanceof RuntimeException)
+            {
+                throw (RuntimeException)e;
+            }
+            throw new RuntimeException(e);
+        }
+    }
+
+    /**
+     * Stop the full query log leaving behind any generated files.
+     */
+    public synchronized void stop()
+    {
+        try
+        {
+            logger.info("Deactivation of full query log requested.");
+            if (binLog != null)
+            {
+                logger.info("Stopping full query log");
+                binLog.stop();
+                binLog = null;
+            }
+        }
+        catch (Exception e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
+    /**
+     * Check whether the full query log is enabled.
+     * @return true if records are recorded and false otherwise.
+     */
+    public boolean enabled()
+    {
+        return binLog != null;
+    }
+
+    /**
+     * This is potentially lossy, but it's not super critical as we will always generally know
+     * when this is happening and roughly how bad it is.
+     */
+    private void logDroppedSample()
+    {
+        droppedSamplesSinceLastLog.incrementAndGet();
+        if (droppedSamplesStatement.warn(new Object[] {droppedSamplesSinceLastLog.get()}))
+        {
+            droppedSamplesSinceLastLog.set(0);
+        }
+    }
+
+    /**
+     * Log an invocation of a batch of queries
+     * @param type The type of the batch
+     * @param queries CQL text of the queries
+     * @param values Values to bind to as parameters for the queries
+     * @param queryOptions Options associated with the query invocation
+     * @param batchTimeMillis Approximate time in milliseconds since the epoch since the batch was invoked
+     */
+    public void logBatch(String type, List<String> queries,  List<List<ByteBuffer>> values, QueryOptions queryOptions, long batchTimeMillis)
+    {
+        Preconditions.checkNotNull(type, "type was null");
+        Preconditions.checkNotNull(queries, "queries was null");
+        Preconditions.checkNotNull(values, "value was null");
+        Preconditions.checkNotNull(queryOptions, "queryOptions was null");
+        Preconditions.checkArgument(batchTimeMillis > 0, "batchTimeMillis must be > 0");
+
+        //Don't construct the wrapper if the log is disabled
+        BinLog binLog = this.binLog;
+        if (binLog == null)
+        {
+            return;
+        }
+
+        WeighableMarshallableBatch wrappedBatch = new WeighableMarshallableBatch(type, queries, values, queryOptions, batchTimeMillis);
+        logRecord(wrappedBatch, binLog);
+    }
+
+    void logRecord(AbstractWeighableMarshallable record, BinLog binLog)
+    {
+
+        boolean putInQueue = false;
+        try
+        {
+            if (blocking)
+            {
+                try
+                {
+                    binLog.put(record);
+                    putInQueue = true;
+                }
+                catch (InterruptedException e)
+                {
+                    throw new RuntimeException(e);
+                }
+            }
+            else
+            {
+                if (!binLog.offer(record))
+                {
+                    logDroppedSample();
+                }
+                else
+                {
+                    putInQueue = true;
+                }
+            }
+        }
+        finally
+        {
+            if (!putInQueue)
+            {
+                record.release();
+            }
+        }
+    }
+
+    /**
+     * Log a single CQL query
+     * @param query CQL query text
+     * @param queryOptions Options associated with the query invocation
+     * @param queryTimeMillis Approximate time in milliseconds since the epoch since the batch was invoked
+     */
+    public void logQuery(String query, QueryOptions queryOptions, long queryTimeMillis)
+    {
+        Preconditions.checkNotNull(query, "query was null");
+        Preconditions.checkNotNull(queryOptions, "queryOptions was null");
+        Preconditions.checkArgument(queryTimeMillis > 0, "queryTimeMillis must be > 0");
+
+        //Don't construct the wrapper if the log is disabled
+        BinLog binLog = this.binLog;
+        if (binLog == null)
+        {
+            return;
+        }
+
+        WeighableMarshallableQuery wrappedQuery = new WeighableMarshallableQuery(query, queryOptions, queryTimeMillis);
+        logRecord(wrappedQuery, binLog);
+    }
+
+    private static abstract class AbstractWeighableMarshallable extends BinLog.ReleaseableWriteMarshallable implements WeightedQueue.Weighable
+    {
+        private final ByteBuf queryOptionsBuffer;
+        private final long timeMillis;
+        private final int protocolVersion;
+
+        private AbstractWeighableMarshallable(QueryOptions queryOptions, long timeMillis)
+        {
+            this.timeMillis = timeMillis;
+            ProtocolVersion version = queryOptions.getProtocolVersion();
+            this.protocolVersion = version.asInt();
+            int optionsSize = QueryOptions.codec.encodedSize(queryOptions, version);
+            queryOptionsBuffer = CBUtil.allocator.buffer(optionsSize, optionsSize);
+            /*
+             * Struggled with what tradeoff to make in terms of query options which is potentially large and complicated
+             * There is tension between low garbage production (or allocator overhead), small working set size, and CPU overhead reserializing the
+             * query options into binary format.
+             *
+             * I went with the lowest risk most predictable option which is allocator overhead and CPU overhead
+             * rather then keep the original query message around so I could just serialize that as a memcpy. It's more
+             * instructions when turned on, but it doesn't change memory footprint quite as much and it's more pay for what you use
+             * in terms of query volume. The CPU overhead is spread out across producers so we should at least get
+             * some scaling.
+             *
+             */
+            boolean success = false;
+            try
+            {
+                QueryOptions.codec.encode(queryOptions, queryOptionsBuffer, version);
+                success = true;
+            }
+            finally
+            {
+                if (!success)
+                {
+                    queryOptionsBuffer.release();
+                }
+            }
+        }
+
+        @Override
+        public void writeMarshallable(WireOut wire)
+        {
+            wire.write("protocol-version").int32(protocolVersion);
+            wire.write("query-options").bytes(BytesStore.wrap(queryOptionsBuffer.nioBuffer()));
+            wire.write("query-time").int64(timeMillis);
+        }
+
+        @Override
+        public void release()
+        {
+            queryOptionsBuffer.release();
+        }
+
+        //8-bytes for protocol version (assume alignment cost), 8-byte timestamp, 8-byte object header + other contents
+        @Override
+        public int weight()
+        {
+            return 8 + 8 + OBJECT_HEADER_SIZE + EMPTY_BYTEBUF_SIZE + queryOptionsBuffer.capacity();
+        }
+    }
+
+    static class WeighableMarshallableBatch extends AbstractWeighableMarshallable
+    {
+        private final int weight;
+        private final String batchType;
+        private final List<String> queries;
+        private final List<List<ByteBuffer>> values;
+
+        public WeighableMarshallableBatch(String batchType, List<String> queries, List<List<ByteBuffer>> values, QueryOptions queryOptions, long batchTimeMillis)
+        {
+           super(queryOptions, batchTimeMillis);
+           this.queries = queries;
+           this.values = values;
+           this.batchType = batchType;
+           boolean success = false;
+           try
+           {
+
+               //weight, batch type, queries, values
+               int weightTemp = 8 + EMPTY_LIST_SIZE + EMPTY_LIST_SIZE;
+               for (int ii = 0; ii < queries.size(); ii++)
+               {
+                   weightTemp += ObjectSizes.sizeOf(queries.get(ii));
+               }
+
+               weightTemp += EMPTY_LIST_SIZE * values.size();
+               for (int ii = 0; ii < values.size(); ii++)
+               {
+                   List<ByteBuffer> sublist = values.get(ii);
+                   weightTemp += EMPTY_BYTEBUFFER_SIZE * sublist.size();
+                   for (int zz = 0; zz < sublist.size(); zz++)
+                   {
+                       weightTemp += sublist.get(zz).capacity();
+                   }
+               }
+               weightTemp += super.weight();
+               weightTemp += ObjectSizes.sizeOf(batchType);
+               weight = weightTemp;
+               success = true;
+           }
+           finally
+           {
+               if (!success)
+               {
+                   release();
+               }
+           }
+        }
+
+        @Override
+        public void writeMarshallable(WireOut wire)
+        {
+            wire.write("type").text("batch");
+            super.writeMarshallable(wire);
+            wire.write("batch-type").text(batchType);
+            ValueOut valueOut = wire.write("queries");
+            valueOut.int32(queries.size());
+            for (String query : queries)
+            {
+                valueOut.text(query);
+            }
+            valueOut = wire.write("values");
+            valueOut.int32(values.size());
+            for (List<ByteBuffer> subValues : values)
+            {
+                valueOut.int32(subValues.size());
+                for (ByteBuffer value : subValues)
+                {
+                    valueOut.bytes(BytesStore.wrap(value));
+                }
+            }
+        }
+
+        @Override
+        public int weight()
+        {
+            return weight;
+        }
+
+    }
+
+    static class WeighableMarshallableQuery extends AbstractWeighableMarshallable
+    {
+        private final String query;
+
+        public WeighableMarshallableQuery(String query, QueryOptions queryOptions, long queryTimeMillis)
+        {
+            super(queryOptions, queryTimeMillis);
+            this.query = query;
+        }
+
+        @Override
+        public void writeMarshallable(WireOut wire)
+        {
+            wire.write("type").text("single");
+            super.writeMarshallable(wire);
+            wire.write("query").text(query);
+        }
+
+        @Override
+        public int weight()
+        {
+            return Ints.checkedCast(ObjectSizes.sizeOf(query)) + super.weight();
+        }
+    }
+
+    static Throwable cleanDirectory(File directory, Throwable accumulate)
+    {
+        if (!directory.exists())
+        {
+            return Throwables.merge(accumulate, new RuntimeException(String.format("%s does not exists")));
+        }
+        if (!directory.isDirectory())
+        {
+            return Throwables.merge(accumulate, new RuntimeException(String.format("%s is not a directory", directory)));
+        }
+        for (File f : directory.listFiles())
+        {
+            accumulate = deleteRecursively(f, accumulate);
+        }
+        if (accumulate instanceof FSError)
+        {
+            FileUtils.handleFSError((FSError)accumulate);
+        }
+        return accumulate;
+    }
+
+    private static Throwable deleteRecursively(File fileOrDirectory, Throwable accumulate)
+    {
+        if (fileOrDirectory.isDirectory())
+        {
+            for (File f : fileOrDirectory.listFiles())
+            {
+                accumulate = FileUtils.deleteWithConfirm(f, true, accumulate);
+            }
+        }
+        return FileUtils.deleteWithConfirm(fileOrDirectory, true , accumulate);
+    }
+}

src/java/org/apache/cassandra/service/StorageProxy.java

@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.lang.management.ManagementFactory;
 import java.net.InetAddress;
 import java.nio.ByteBuffer;
+import java.nio.file.Paths;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -29,6 +30,7 @@ import java.util.concurrent.atomic.AtomicLong;
 import javax.management.MBeanServer;
 import javax.management.ObjectName;
 
+import com.google.common.base.Preconditions;
 import com.google.common.base.Predicate;
 import com.google.common.cache.CacheLoader;
 import com.google.common.collect.*;
@@ -51,6 +53,7 @@ import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.filter.DataLimits;
 import org.apache.cassandra.db.filter.TombstoneOverwhelmingException;
+import org.apache.cassandra.db.fullquerylog.FullQueryLogger;
 import org.apache.cassandra.db.partitions.*;
 import org.apache.cassandra.db.rows.RowIterator;
 import org.apache.cassandra.db.view.ViewUtils;
@@ -2860,6 +2863,26 @@ public class StorageProxy implements StorageProxyMBean
         return String.format("Updating ideal consistency level new value: %s old value %s", newCL, original.toString());
     }
 
+    @Override
+    public void configureFullQueryLogger(String path, String rollCycle, boolean blocking, int maxQueueWeight, long maxLogSize)
+    {
+        path = path != null ? path : DatabaseDescriptor.getFullQueryLogPath();
+        Preconditions.checkNotNull(path, "cassandra.yaml did not set full_query_log_dir and not set as parameter");
+        FullQueryLogger.instance.configure(Paths.get(path), rollCycle, blocking, maxQueueWeight, maxLogSize);
+    }
+
+    @Override
+    public void resetFullQueryLogger()
+    {
+        FullQueryLogger.instance.reset(DatabaseDescriptor.getFullQueryLogPath());
+    }
+
+    @Override
+    public void stopFullQueryLogger()
+    {
+        FullQueryLogger.instance.stop();
+    }
+
     public int getOtcBacklogExpirationInterval() {
         return DatabaseDescriptor.getOtcBacklogExpirationInterval();
     }

src/java/org/apache/cassandra/service/StorageProxyMBean.java

@@ -71,4 +71,25 @@ public interface StorageProxyMBean
 
     public String getIdealConsistencyLevel();
     public String setIdealConsistencyLevel(String cl);
+
+    /**
+     * Start the fully query logger.
+     * @param path Path where the full query log will be stored. If null cassandra.yaml value is used.
+     * @param rollCycle How often to create a new file for query data (MINUTELY, DAILY, HOURLY)
+     * @param blocking Whether threads submitting queries to the query log should block if they can't be drained to the filesystem or alternatively drops samples and log
+     * @param maxQueueWeight How many bytes of query data to queue before blocking or dropping samples
+     * @param maxLogSize How many bytes of log data to store before dropping segments. Might not be respected if a log file hasn't rolled so it can be deleted.
+     */
+    public void configureFullQueryLogger(String path, String rollCycle, boolean blocking, int maxQueueWeight, long maxLogSize);
+
+    /**
+     * Disable the full query logger if it is enabled.
+     * Also delete any generated files in the last used full query log path as well as the one configure in cassandra.yaml
+     */
+    public void resetFullQueryLogger();
+
+    /**
+     * Stop logging queries but leave any generated files on disk.
+     */
+    public void stopFullQueryLogger();
 }

src/java/org/apache/cassandra/tools/FullQueryLogTool.java

@@ -0,0 +1,93 @@
+/*
+ * 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.tools;
+
+import java.util.List;
+
+import com.google.common.base.Throwables;
+
+import io.airlift.airline.Cli;
+import io.airlift.airline.Help;
+import io.airlift.airline.ParseArgumentsMissingException;
+import io.airlift.airline.ParseArgumentsUnexpectedException;
+import io.airlift.airline.ParseCommandMissingException;
+import io.airlift.airline.ParseCommandUnrecognizedException;
+import io.airlift.airline.ParseOptionConversionException;
+import io.airlift.airline.ParseOptionMissingException;
+import io.airlift.airline.ParseOptionMissingValueException;
+import org.apache.cassandra.tools.fqltool.Dump;
+
+import static com.google.common.base.Throwables.getStackTraceAsString;
+import static com.google.common.collect.Lists.newArrayList;
+
+public class FullQueryLogTool
+{
+    public static void main(String... args)
+    {
+        List<Class<? extends Runnable>> commands = newArrayList(
+                Help.class,
+                Dump.class
+        );
+
+        Cli.CliBuilder<Runnable> builder = Cli.builder("fqltool");
+
+        builder.withDescription("Manipulate the contents of full query log files")
+                 .withDefaultCommand(Help.class)
+                 .withCommands(commands);
+
+        Cli<Runnable> parser = builder.build();
+
+        int status = 0;
+        try
+        {
+            parser.parse(args).run();
+        } catch (IllegalArgumentException |
+                IllegalStateException |
+                ParseArgumentsMissingException |
+                ParseArgumentsUnexpectedException |
+                ParseOptionConversionException |
+                ParseOptionMissingException |
+                ParseOptionMissingValueException |
+                ParseCommandMissingException |
+                ParseCommandUnrecognizedException e)
+        {
+            badUse(e);
+            status = 1;
+        } catch (Throwable throwable)
+        {
+            err(Throwables.getRootCause(throwable));
+            status = 2;
+        }
+
+        System.exit(status);
+    }
+
+    private static void badUse(Exception e)
+    {
+        System.out.println("fqltool: " + e.getMessage());
+        System.out.println("See 'fqltool help' or 'fqltool help <command>'.");
+    }
+
+    private static void err(Throwable e)
+    {
+        System.err.println("error: " + e.getMessage());
+        System.err.println("-- StackTrace --");
+        System.err.println(getStackTraceAsString(e));
+    }
+}

src/java/org/apache/cassandra/tools/NodeTool.java

@@ -76,6 +76,8 @@ public class NodeTool
                 EnableGossip.class,
                 DisableGossip.class,
                 EnableHandoff.class,
+                EnableFullQueryLog.class,
+                DisableFullQueryLog.class,
                 GcStats.class,
                 GetBatchlogReplayTrottle.class,
                 GetCompactionThreshold.class,
@@ -100,6 +102,7 @@ public class NodeTool
                 Refresh.class,
                 RemoveNode.class,
                 Assassinate.class,
+                ResetFullQueryLog.class,
                 Repair.class,
                 RepairAdmin.class,
                 ReplayBatchlog.class,

src/java/org/apache/cassandra/tools/fqltool/Dump.java

@@ -0,0 +1,265 @@
+/*
+ * 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.tools.fqltool;
+
+import java.io.File;
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import io.airlift.airline.Arguments;
+import io.airlift.airline.Command;
+import io.airlift.airline.Option;
+import io.netty.buffer.Unpooled;
+import net.openhft.chronicle.bytes.Bytes;
+import net.openhft.chronicle.queue.ChronicleQueue;
+import net.openhft.chronicle.queue.ChronicleQueueBuilder;
+import net.openhft.chronicle.queue.ExcerptTailer;
+import net.openhft.chronicle.queue.RollCycles;
+import net.openhft.chronicle.threads.Pauser;
+import net.openhft.chronicle.wire.ReadMarshallable;
+import net.openhft.chronicle.wire.ValueIn;
+import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.transport.ProtocolVersion;
+
+/**
+ * Dump the contents of a list of paths containing full query logs
+ */
+@Command(name = "dump", description = "Dump the contents of a full query log")
+public class Dump implements Runnable
+{
+    static final char[] HEXI_DECIMAL = "0123456789ABCDEF".toCharArray();
+
+    @Arguments(usage = "<path1> [<path2>...<pathN>]", description = "Path containing the full query logs to dump.", required = true)
+    private List<String> arguments = new ArrayList<>();
+
+    @Option(title = "roll_cycle", name = {"--roll-cycle"}, description = "How often to roll the log file was rolled. May be necessary for Chronicle to correctly parse file names. (MINUTELY, HOURLY, DAILY). Default HOURLY.")
+    private String rollCycle = "HOURLY";
+
+    @Option(title = "follow", name = {"--follow"}, description = "Upon reacahing the end of the log continue indefinitely waiting for more records")
+    private boolean follow = false;
+
+    @Override
+    public void run()
+    {
+        dump(arguments, rollCycle, follow);
+    }
+
+    public static void dump(List<String> arguments, String rollCycle, boolean follow)
+    {
+        StringBuilder sb = new StringBuilder();
+        ReadMarshallable reader = wireIn -> {
+            sb.setLength(0);
+            String type = wireIn.read("type").text();
+            sb.append("Type: ").append(type).append(System.lineSeparator());
+            int protocolVersion = wireIn.read("protocol-version").int32();
+            sb.append("Protocol version: ").append(protocolVersion).append(System.lineSeparator());
+            QueryOptions options = QueryOptions.codec.decode(Unpooled.wrappedBuffer(wireIn.read("query-options").bytesStore().toTemporaryDirectByteBuffer()), ProtocolVersion.decode(protocolVersion));
+            sb.append("Query time: ").append(wireIn.read("query-time").int64()).append(System.lineSeparator());
+            if (type.equals("single"))
+            {
+                sb.append("Query: ").append(wireIn.read("query").text()).append(System.lineSeparator());
+                List<ByteBuffer> values = options.getValues() != null ? options.getValues() : Collections.EMPTY_LIST;
+                sb.append("Values: ").append(System.lineSeparator());
+                valuesToStringBuilder(values, sb);
+            }
+            else
+            {
+                sb.append("Batch type: ").append(wireIn.read("batch-type").text()).append(System.lineSeparator());
+                ValueIn in = wireIn.read("queries");
+                int numQueries = in.int32();
+                List<String> queries = new ArrayList<>();
+                for (int ii = 0; ii < numQueries; ii++)
+                {
+                    queries.add(in.text());
+                }
+                in = wireIn.read("values");
+                int numValues = in.int32();
+                List<List<ByteBuffer>> values = new ArrayList<>();
+                for (int ii = 0; ii < numValues; ii++)
+                {
+                    List<ByteBuffer> subValues = new ArrayList<>();
+                    values.add(subValues);
+                    int numSubValues = in.int32();
+                    for (int zz = 0; zz < numSubValues; zz++)
+                    {
+                        subValues.add(ByteBuffer.wrap(in.bytes()));
+                    }
+                    sb.append("Query: ").append(queries.get(ii)).append(System.lineSeparator());
+                    sb.append("Values: ").append(System.lineSeparator());
+                    valuesToStringBuilder(subValues, sb);
+                }
+            }
+            sb.append(System.lineSeparator());
+            System.out.print(sb.toString());
+            System.out.flush();
+        };
+
+        //Backoff strategy for spinning on the queue, not aggressive at all as this doesn't need to be low latency
+        Pauser pauser = Pauser.millis(100);
+        List<ChronicleQueue> queues = arguments.stream().distinct().map(path -> ChronicleQueueBuilder.single(new File(path)).rollCycle(RollCycles.valueOf(rollCycle)).build()).collect(Collectors.toList());
+        List<ExcerptTailer> tailers = queues.stream().map(ChronicleQueue::createTailer).collect(Collectors.toList());
+        boolean hadWork = true;
+        while (hadWork)
+        {
+            hadWork = false;
+            for (ExcerptTailer tailer : tailers)
+            {
+                while (tailer.readDocument(reader))
+                {
+                    hadWork = true;
+                }
+            }
+
+            if (follow)
+            {
+                if (!hadWork)
+                {
+                    //Chronicle queue doesn't support blocking so use this backoff strategy
+                    pauser.pause();
+                }
+                //Don't terminate the loop even if there wasn't work
+                hadWork = true;
+            }
+        }
+    }
+
+    private static void valuesToStringBuilder(List<ByteBuffer> values, StringBuilder sb)
+    {
+        boolean first = true;
+        for (ByteBuffer value : values)
+        {
+            Bytes bytes = Bytes.wrapForRead(value);
+            long maxLength2 = Math.min(1024, bytes.readLimit() - bytes.readPosition());
+            toHexString(bytes, bytes.readPosition(), maxLength2, sb);
+            if (maxLength2 < bytes.readLimit() - bytes.readPosition())
+            {
+                sb.append("... truncated").append(System.lineSeparator());
+            }
+
+            if (first)
+            {
+                first = false;
+            }
+            else
+            {
+                sb.append("-----").append(System.lineSeparator());
+            }
+        }
+    }
+
+    //This is from net.openhft.chronicle.bytes, need to pass in the StringBuilder so had to copy
+    /*
+     * Copyright 2016 higherfrequencytrading.com
+     *
+     * Licensed 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.
+     */
+    /**
+     * display the hex data of {@link Bytes} from the position() to the limit()
+     *
+     * @param bytes the buffer you wish to toString()
+     * @return hex representation of the buffer, from example [0D ,OA, FF]
+     */
+    public static String toHexString(final Bytes bytes, long offset, long len, StringBuilder builder)
+    throws BufferUnderflowException
+    {
+        if (len == 0)
+            return "";
+
+        int width = 16;
+        int[] lastLine = new int[width];
+        String sep = "";
+        long position = bytes.readPosition();
+        long limit = bytes.readLimit();
+
+        try {
+            bytes.readPositionRemaining(offset, len);
+
+            long start = offset / width * width;
+            long end = (offset + len + width - 1) / width * width;
+            for (long i = start; i < end; i += width) {
+                // check for duplicate rows
+                if (i + width < end) {
+                    boolean same = true;
+
+                    for (int j = 0; j < width && i + j < offset + len; j++) {
+                        int ch = bytes.readUnsignedByte(i + j);
+                        same &= (ch == lastLine[j]);
+                        lastLine[j] = ch;
+                    }
+                    if (i > start && same) {
+                        sep = "........\n";
+                        continue;
+                    }
+                }
+                builder.append(sep);
+                sep = "";
+                String str = Long.toHexString(i);
+                for (int j = str.length(); j < 8; j++)
+                    builder.append('0');
+                builder.append(str);
+                for (int j = 0; j < width; j++) {
+                    if (j == width / 2)
+                        builder.append(' ');
+                    if (i + j < offset || i + j >= offset + len) {
+                        builder.append("   ");
+
+                    } else {
+                        builder.append(' ');
+                        int ch = bytes.readUnsignedByte(i + j);
+                        builder.append(HEXI_DECIMAL[ch >> 4]);
+                        builder.append(HEXI_DECIMAL[ch & 15]);
+                    }
+                }
+                builder.append(' ');
+                for (int j = 0; j < width; j++) {
+                    if (j == width / 2)
+                        builder.append(' ');
+                    if (i + j < offset || i + j >= offset + len) {
+                        builder.append(' ');
+
+                    } else {
+                        int ch = bytes.readUnsignedByte(i + j);
+                        if (ch < ' ' || ch > 126)
+                            ch = '\u00B7';
+                        builder.append((char) ch);
+                    }
+                }
+                builder.append("\n");
+            }
+            return builder.toString();
+        } finally {
+            bytes.readLimit(limit);
+            bytes.readPosition(position);
+        }
+    }
+}

src/java/org/apache/cassandra/tools/nodetool/DisableFullQueryLog.java

@@ -0,0 +1,33 @@
+/*
+ * 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.tools.nodetool;
+
+import io.airlift.airline.Command;
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
+
+@Command(name = "disablefullquerylog", description = "Disable the full query log")
+public class DisableFullQueryLog extends NodeToolCmd
+{
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        probe.getSpProxy().stopFullQueryLogger();
+    }
+}

src/java/org/apache/cassandra/tools/nodetool/EnableFullQueryLog.java

@@ -0,0 +1,49 @@
+/*
+ * 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.tools.nodetool;
+
+import io.airlift.airline.Command;
+import io.airlift.airline.Option;
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
+
+@Command(name = "enablefullquerylog", description = "Enable full query logging")
+public class EnableFullQueryLog extends NodeToolCmd
+{
+    @Option(title = "roll_cycle", name = {"--roll-cycle"}, description = "How often to roll the log file (MINUTELY, HOURLY, DAILY). Default HOURLY.")
+    private String rollCycle = "HOURLY";
+
+    @Option(title = "blocking", name = {"--blocking"}, description = "If the queue is full whether to block producers or drop samples. Default true.")
+    private boolean blocking = true;
+
+    @Option(title = "max_queue_weight", name = {"--max-queue-weight"}, description = "Maximum number of bytes of query data to queue to disk before blocking or dropping samples. Default 256 megabytes.")
+    private int maxQueueWeight = 256 * 1024 * 1024;
+
+    @Option(title = "max_log_size", name = {"--max-log-size"}, description = "How many bytes of log data to store before dropping segments. Might not be respected if a log file hasn't rolled so it can be deleted. Default 16 gigabytes.")
+    private long maxLogSize = 16L * 1024L * 1024L * 1024L;
+
+    @Option(title = "path", name = {"--path"}, description = "Path to store the full query log at. Will have it's contents recursively deleted. If not set the value from cassandra.yaml will be used.")
+    private String path = null;
+
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        probe.getSpProxy().configureFullQueryLogger(path, rollCycle, blocking, maxQueueWeight, maxLogSize);
+    }
+}

src/java/org/apache/cassandra/tools/nodetool/ResetFullQueryLog.java

@@ -0,0 +1,33 @@
+/*
+ * 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.tools.nodetool;
+
+import io.airlift.airline.Command;
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
+
+@Command(name = "resetfullquerylog", description = "Stop the full query log and clean files in the configured full query log directory from cassandra.yaml as well as JMX")
+public class ResetFullQueryLog extends NodeToolCmd
+{
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        probe.getSpProxy().resetFullQueryLogger();
+    }
+}

src/java/org/apache/cassandra/transport/messages/BatchMessage.java

@@ -19,6 +19,7 @@ package org.apache.cassandra.transport.messages;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
 
@@ -29,6 +30,7 @@ import org.apache.cassandra.cql3.*;
 import org.apache.cassandra.cql3.statements.BatchStatement;
 import org.apache.cassandra.cql3.statements.ModificationStatement;
 import org.apache.cassandra.cql3.statements.ParsedStatement;
+import org.apache.cassandra.db.fullquerylog.FullQueryLogger;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.exceptions.PreparedQueryNotFoundException;
 import org.apache.cassandra.service.ClientState;
@@ -174,20 +176,30 @@ public class BatchMessage extends Message.Request
 
             QueryHandler handler = ClientState.getCQLQueryHandler();
             List<ParsedStatement.Prepared> prepared = new ArrayList<>(queryOrIdList.size());
+            boolean fullQueryLogEnabled = FullQueryLogger.instance.enabled();
+            List<String> queryStrings = fullQueryLogEnabled ? new ArrayList<>(queryOrIdList.size()) : Collections.EMPTY_LIST;
             for (int i = 0; i < queryOrIdList.size(); i++)
             {
                 Object query = queryOrIdList.get(i);
+                String queryString;
                 ParsedStatement.Prepared p;
                 if (query instanceof String)
                 {
                     p = QueryProcessor.parseStatement((String)query,
                                                       state.getClientState().cloneWithKeyspaceIfSet(options.getKeyspace()));
+                    queryString = (String)query;
                 }
                 else
                 {
                     p = handler.getPrepared((MD5Digest)query);
                     if (p == null)
                         throw new PreparedQueryNotFoundException((MD5Digest)query);
+                    queryString = p.rawCQLStatement;
+                }
+
+                if (fullQueryLogEnabled)
+                {
+                    queryStrings.add(queryString);
                 }
 
                 List<ByteBuffer> queryValues = values.get(i);
@@ -215,7 +227,16 @@ public class BatchMessage extends Message.Request
             // Note: It's ok at this point to pass a bogus value for the number of bound terms in the BatchState ctor
             // (and no value would be really correct, so we prefer passing a clearly wrong one).
             BatchStatement batch = new BatchStatement(-1, batchType, statements, Attributes.none());
+            long fqlTime = 0;
+            if (fullQueryLogEnabled)
+            {
+                fqlTime = System.currentTimeMillis();
+            }
             Message.Response response = handler.processBatch(batch, state, batchOptions, getCustomPayload(), queryStartNanoTime);
+            if (fullQueryLogEnabled)
+            {
+                FullQueryLogger.instance.logBatch(batchType.name(), queryStrings, values, options, fqlTime);
+            }
 
             if (tracingId != null)
                 response.setTracingId(tracingId);

src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java

@@ -31,6 +31,7 @@ import org.apache.cassandra.cql3.statements.BatchStatement;
 import org.apache.cassandra.cql3.statements.ModificationStatement;
 import org.apache.cassandra.cql3.statements.ParsedStatement;
 import org.apache.cassandra.cql3.statements.UpdateStatement;
+import org.apache.cassandra.db.fullquerylog.FullQueryLogger;
 import org.apache.cassandra.exceptions.PreparedQueryNotFoundException;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.QueryState;
@@ -162,7 +163,17 @@ public class ExecuteMessage extends Message.Request
             // Some custom QueryHandlers are interested by the bound names. We provide them this information
             // by wrapping the QueryOptions.
             QueryOptions queryOptions = QueryOptions.addColumnSpecifications(options, prepared.boundNames);
+            boolean fqlEnabled = FullQueryLogger.instance.enabled();
+            long fqlTime = 0;
+            if (fqlEnabled)
+            {
+                fqlTime = System.currentTimeMillis();
+            }
             Message.Response response = handler.processPrepared(statement, state, queryOptions, getCustomPayload(), queryStartNanoTime);
+            if (fqlEnabled)
+            {
+                FullQueryLogger.instance.logQuery(prepared.rawCQLStatement, options, fqlTime);
+            }
 
             if (response instanceof ResultMessage.Rows)
             {

src/java/org/apache/cassandra/transport/messages/QueryMessage.java

@@ -23,6 +23,7 @@ import com.google.common.collect.ImmutableMap;
 
 import io.netty.buffer.ByteBuf;
 import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.db.fullquerylog.FullQueryLogger;
 import org.apache.cassandra.exceptions.RequestExecutionException;
 import org.apache.cassandra.exceptions.RequestValidationException;
 import org.apache.cassandra.service.ClientState;
@@ -113,7 +114,17 @@ public class QueryMessage extends Message.Request
                 Tracing.instance.begin("Execute CQL3 query", state.getClientAddress(), builder.build());
             }
 
+            boolean fqlEnabled = FullQueryLogger.instance.enabled();
+            long fqlTime = 0;
+            if (fqlEnabled)
+            {
+                fqlTime = System.currentTimeMillis();
+            }
             Message.Response response = ClientState.getCQLQueryHandler().process(query, state, options, getCustomPayload(), queryStartNanoTime);
+            if (fqlEnabled)
+            {
+                FullQueryLogger.instance.logQuery(query, options, fqlTime);
+            }
             if (options.skipMetadata() && response instanceof ResultMessage.Rows)
                 ((ResultMessage.Rows)response).result.metadata.setSkipMetadata();
 

src/java/org/apache/cassandra/utils/ObjectSizes.java

@@ -138,6 +138,7 @@ public class ObjectSizes
      * @param str String to calculate memory size of
      * @return Total in-memory size of the String
      */
+    //@TODO hard coding this to 2 isn't necessarily correct in Java 9
     public static long sizeOf(String str)
     {
         return STRING_EMPTY_SIZE + sizeOfArray(str.length(), 2);

src/java/org/apache/cassandra/utils/binlog/BinLog.java

@@ -0,0 +1,277 @@
+/*
+ * 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.utils.binlog;
+
+import java.io.File;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import net.openhft.chronicle.queue.ChronicleQueue;
+import net.openhft.chronicle.queue.ChronicleQueueBuilder;
+import net.openhft.chronicle.queue.ExcerptAppender;
+import net.openhft.chronicle.queue.RollCycle;
+import net.openhft.chronicle.queue.impl.StoreFileListener;
+import net.openhft.chronicle.wire.WireOut;
+import net.openhft.chronicle.wire.WriteMarshallable;
+import org.apache.cassandra.concurrent.NamedThreadFactory;
+import org.apache.cassandra.utils.concurrent.WeightedQueue;
+
+/**
+ * Bin log is a is quick and dirty binary log that is kind of a NIH version of binary logging with a traditional logging
+ * framework. It's goal is good enough performance, predictable footprint, simplicity in terms of implementation and configuration
+ * and most importantly minimal impact on producers of log records.
+ *
+ * Performance safety is accomplished by feeding items to the binary log using a weighted queue and dropping records if the binary log falls
+ * sufficiently far behind.
+ *
+ * Simplicity and good enough perforamance is achieved by using a single log writing thread as well as Chronicle Queue
+ * to handle writing the log, making it available for readers, as well as log rolling.
+ *
+ */
+public class BinLog implements Runnable, StoreFileListener
+{
+    private static final Logger logger = LoggerFactory.getLogger(BinLog.class);
+
+    private ChronicleQueue queue;
+    private ExcerptAppender appender;
+    @VisibleForTesting
+    Thread binLogThread = new NamedThreadFactory("Binary Log thread").newThread(this);
+    final WeightedQueue<ReleaseableWriteMarshallable> sampleQueue;
+    private final long maxLogSize;
+
+    /**
+     * The files in the chronicle queue that have already rolled
+     */
+    private Queue<File> chronicleStoreFiles = new ConcurrentLinkedQueue<>();
+
+    /**
+     * The number of bytes in store files that have already rolled
+     */
+    private long bytesInStoreFiles;
+
+    private static final ReleaseableWriteMarshallable NO_OP = new ReleaseableWriteMarshallable()
+    {
+        @Override
+        public void writeMarshallable(WireOut wire)
+        {
+        }
+
+        @Override
+        public void release()
+        {
+        }
+    };
+
+    private volatile boolean shouldContinue = true;
+
+    /**
+     *
+     * @param path Path to store the BinLog. Can't be shared with anything else.
+     * @param rollCycle How often to roll the log file so it can potentially be deleted
+     * @param maxQueueWeight Maximum weight of in memory queue for records waiting to be written to the file before blocking or dropping
+     * @param maxLogSize Maximum size of the rolled files to retain on disk before deleting the oldest file
+     */
+    public BinLog(Path path, RollCycle rollCycle, int maxQueueWeight, long maxLogSize)
+    {
+        Preconditions.checkNotNull(path, "path was null");
+        Preconditions.checkNotNull(rollCycle, "rollCycle was null");
+        Preconditions.checkArgument(maxQueueWeight > 0, "maxQueueWeight must be > 0");
+        Preconditions.checkArgument(maxLogSize > 0, "maxLogSize must be > 0");
+        ChronicleQueueBuilder builder = ChronicleQueueBuilder.single(path.toFile());
+        builder.rollCycle(rollCycle);
+        builder.storeFileListener(this);
+        queue = builder.build();
+        appender = queue.acquireAppender();
+        sampleQueue = new WeightedQueue<>(maxQueueWeight);
+        this.maxLogSize = maxLogSize;
+    }
+
+    /**
+     * Start the consumer thread that writes log records. Can only be done once.
+     */
+    public void start()
+    {
+        if (!shouldContinue)
+        {
+            throw new IllegalStateException("Can't reuse stopped BinLog");
+        }
+        binLogThread.start();
+    }
+
+    /**
+     * Stop the consumer thread that writes log records. Can be called multiple times.
+     * @throws InterruptedException
+     */
+    public synchronized void stop() throws InterruptedException
+    {
+        if (!shouldContinue)
+        {
+            return;
+        }
+
+        shouldContinue = false;
+        sampleQueue.put(NO_OP);
+        binLogThread.join();
+        appender = null;
+        queue = null;
+    }
+
+    /**
+     * Offer a record to the log. If the in memory queue is full the record will be dropped and offer will return false.
+     * @param record The record to write to the log
+     * @return true if the record was queued and false otherwise
+     */
+    public boolean offer(ReleaseableWriteMarshallable record)
+    {
+        if (!shouldContinue)
+        {
+            return false;
+        }
+
+        return sampleQueue.offer(record);
+    }
+
+    /**
+     * Put a record into the log. If the in memory queue is full the putting thread will be blocked until there is space or it is interrupted.
+     * @param record The record to write to the log
+     * @throws InterruptedException
+     */
+    public void put(ReleaseableWriteMarshallable record) throws InterruptedException
+    {
+        if (!shouldContinue)
+        {
+            return;
+        }
+
+        //Resolve potential deadlock at shutdown when queue is full
+        while (shouldContinue)
+        {
+            if (sampleQueue.offer(record, 1, TimeUnit.SECONDS))
+            {
+                return;
+            }
+        }
+    }
+
+    private void processTasks(List<ReleaseableWriteMarshallable> tasks)
+    {
+        for (int ii = 0; ii < tasks.size(); ii++)
+        {
+            WriteMarshallable t = tasks.get(ii);
+            //Don't write an empty document
+            if (t == NO_OP)
+            {
+                continue;
+            }
+
+            appender.writeDocument(t);
+        }
+    }
+
+    @Override
+    public void run()
+    {
+        List<ReleaseableWriteMarshallable> tasks = new ArrayList<>(16);
+        while (shouldContinue)
+        {
+            try
+            {
+                tasks.clear();
+                ReleaseableWriteMarshallable task = sampleQueue.take();
+                tasks.add(task);
+                sampleQueue.drainTo(tasks, 15);
+
+                processTasks(tasks);
+            }
+            catch (Throwable t)
+            {
+                logger.error("Unexpected exception in binary log thread", t);
+            }
+            finally
+            {
+                for (int ii = 0; ii < tasks.size(); ii++)
+                {
+                    tasks.get(ii).release();
+                }
+            }
+        }
+
+        //Clean up the buffers on thread exit, finalization will check again once this
+        //is no longer reachable ensuring there are no stragglers in the queue.
+        finalize();
+    }
+
+    /**
+     * Track store files as they are added and their storage impact. Delete them if over storage limit.
+     * @param cycle
+     * @param file
+     */
+    public synchronized void onReleased(int cycle, File file)
+    {
+        chronicleStoreFiles.offer(file);
+        //This isn't accurate because the files are sparse, but it's at least pessimistic
+        bytesInStoreFiles += file.length();
+        logger.debug("Chronicle store file {} rolled file size {}", file.getPath(), file.length());
+        while (bytesInStoreFiles > maxLogSize & !chronicleStoreFiles.isEmpty())
+        {
+            File toDelete = chronicleStoreFiles.poll();
+            long toDeleteLength = toDelete.length();
+            if (!toDelete.delete())
+            {
+                logger.error("Failed to delete chronicle store file: {} store file size: {} bytes in store files: {}. " +
+                             "You will need to clean this up manually or reset full query logging.",
+                             toDelete.getPath(), toDeleteLength, bytesInStoreFiles);
+            }
+            else
+            {
+                bytesInStoreFiles -= toDeleteLength;
+                logger.info("Deleted chronicle store file: {} store file size: {} bytes in store files: {} max log size: {}.", file.getPath(), toDeleteLength, bytesInStoreFiles, maxLogSize);
+            }
+        }
+    }
+
+    /**
+     * There is a race where we might not release a buffer, going to let finalization
+     * catch it since it shouldn't happen to a lot of buffers. Only test code would run
+     * into it anyways.
+     */
+    @Override
+    public void finalize()
+    {
+        ReleaseableWriteMarshallable toRelease;
+        while (((toRelease = sampleQueue.poll()) != null))
+        {
+            toRelease.release();
+        }
+    }
+
+    public abstract static class ReleaseableWriteMarshallable implements WriteMarshallable
+    {
+        protected abstract void release();
+    }
+}

src/java/org/apache/cassandra/utils/concurrent/WeightedQueue.java

@@ -0,0 +1,333 @@
+/*
+ * 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.utils.concurrent;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Objects;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Weighted queue is a wrapper around any blocking queue that turns it into a blocking weighted queue. The queue
+ * will weigh each element being added and removed. Adding to the queue is blocked if adding would violate
+ * the weight bound.
+ *
+ * If an element weighs in at larger than the capacity of the queue then exactly one such element will be allowed
+ * into the queue at a time.
+ *
+ * If the weight of an object changes after it is added you are going to have a bad time. Checking weight should be
+ * cheap so memoize expensive to compute weights. If weight throws that can also result in leaked permits so it's
+ * always a good idea to memoize weight so it doesn't throw.
+ *
+ * In the interests of not writing unit tests for methods no one uses there is a lot of UnsupportedOperationException.
+ * If you need them then add them and add proper unit tests to WeightedQueueTest. "Good" tests. 100% coverage including
+ * exception paths and resource leaks.
+ **/
+public class WeightedQueue<T> implements BlockingQueue<T>
+{
+    private static final Logger logger = LoggerFactory.getLogger(WeightedQueue.class);
+    public static final Weigher NATURAL_WEIGHER = (Weigher<Object>) weighable ->
+    {
+        if (weighable instanceof Weighable)
+        {
+            return ((Weighable)weighable).weight();
+        }
+        return 1;
+    };
+
+    private final Weigher<T> weigher;
+    private final BlockingQueue<T> queue;
+    private final int maxWeight;
+    final Semaphore availableWeight;
+
+    public boolean add(T e)
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public boolean offer(T t)
+    {
+        Preconditions.checkNotNull(t);
+        boolean acquired = tryAcquireWeight(t);
+        if (acquired)
+        {
+            boolean offered = false;
+            try
+            {
+                offered = queue.offer(t);
+                return offered;
+            }
+            finally
+            {
+                if (!offered)
+                {
+                    releaseWeight(t);
+                }
+            }
+        }
+        return false;
+    }
+
+    public T remove()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public T poll()
+    {
+        T retval = queue.poll();
+        releaseWeight(retval);
+        return retval;
+    }
+
+    public T element()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public T peek()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public void put(T t) throws InterruptedException
+    {
+        Preconditions.checkNotNull(t);
+        acquireWeight(t, 0, null);
+        boolean put = false;
+        try
+        {
+            queue.put(t);
+            put = true;
+        }
+        finally
+        {
+            if (!put)
+            {
+                releaseWeight(t);
+            }
+        }
+    }
+
+    public boolean offer(T t, long timeout, TimeUnit unit) throws InterruptedException
+    {
+        Preconditions.checkNotNull(t);
+        Preconditions.checkNotNull(unit);
+        boolean acquired = acquireWeight(t, timeout, unit);
+        if (acquired)
+        {
+            boolean offered = false;
+            try
+            {
+                offered = queue.offer(t, timeout, unit);
+                return offered;
+            }
+            finally
+            {
+                if (!offered)
+                {
+                    releaseWeight(t);
+                }
+            }
+        }
+        return false;
+    }
+
+    public T take() throws InterruptedException
+    {
+        T retval = queue.take();
+        releaseWeight(retval);
+        return retval;
+    }
+
+    public T poll(long timeout, TimeUnit unit) throws InterruptedException
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public int remainingCapacity()
+    {
+        throw new UnsupportedOperationException("Seems like a bad idea");
+    }
+
+    public boolean remove(Object o)
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public boolean containsAll(Collection<?> c)
+    {
+        throw new UnsupportedOperationException("Seems like a bad idea");
+    }
+
+    public boolean addAll(Collection<? extends T> c)
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public boolean removeAll(Collection<?> c)
+    {
+        throw new UnsupportedOperationException("Seems like a bad idea");
+    }
+
+    public boolean retainAll(Collection<?> c)
+    {
+        throw new UnsupportedOperationException("Seems like a bad idea");
+    }
+
+    public void clear()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public int size()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public boolean isEmpty()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public boolean contains(Object o)
+    {
+        throw new UnsupportedOperationException("Seems like a bad idea");
+    }
+
+    public Iterator<T> iterator()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public Object[] toArray()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public <T1> T1[] toArray(T1[] a)
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public int drainTo(Collection<? super T> c)
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public int drainTo(Collection<? super T> c, int maxElements)
+    {
+        int count = 0;
+        T o;
+        while(count < maxElements && (o = poll()) != null)
+        {
+            c.add(o);
+            count++;
+        }
+        return count;
+    }
+
+    public interface Weigher<T>
+    {
+        int weigh(T weighable);
+    }
+
+    public interface Weighable
+    {
+        int weight();
+    }
+
+    public WeightedQueue(int maxWeight)
+    {
+        this(maxWeight, new LinkedBlockingQueue<T>(), NATURAL_WEIGHER);
+    }
+
+    public WeightedQueue(int maxWeight, BlockingQueue<T> queue, Weigher<T> weigher)
+    {
+        Preconditions.checkNotNull(queue);
+        Preconditions.checkNotNull(weigher);
+        Preconditions.checkArgument(maxWeight > 0);
+        this.maxWeight = maxWeight;
+        this.queue = queue;
+        this.weigher = weigher;
+        availableWeight = new Semaphore(maxWeight);
+    }
+
+    boolean acquireWeight(T weighable, long timeout, TimeUnit unit) throws InterruptedException
+    {
+        int weight = weigher.weigh(weighable);
+        if (weight < 1)
+        {
+            throw new IllegalArgumentException(String.format("Weighable: \"%s\" had illegal weight %d", Objects.toString(weighable), weight));
+        }
+
+        //Allow exactly one overweight element
+        weight = Math.min(maxWeight, weight);
+
+        if (unit != null)
+        {
+            return availableWeight.tryAcquire(weight, timeout, unit);
+        }
+        else
+        {
+            availableWeight.acquire(weight);
+            return true;
+        }
+    }
+
+    boolean tryAcquireWeight(T weighable)
+    {
+        int weight = weigher.weigh(weighable);
+        if (weight < 1)
+        {
+            throw new IllegalArgumentException(String.format("Weighable: \"%s\" had illegal weight %d", Objects.toString(weighable), weight));
+        }
+
+        //Allow exactly one overweight element
+        weight = Math.min(maxWeight, weight);
+
+        return availableWeight.tryAcquire(weight);
+    }
+
+    void releaseWeight(T weighable)
+    {
+        if (weighable == null)
+        {
+            return;
+        }
+
+        int weight = weigher.weigh(weighable);
+        if (weight < 1)
+        {
+            throw new IllegalArgumentException(String.format("Weighable: \"%s\" had illegal weight %d", Objects.toString(weighable), weight));
+        }
+
+        //Allow exactly one overweight element
+        weight = Math.min(maxWeight, weight);
+
+        availableWeight.release(weight);
+    }
+}

test/unit/org/apache/cassandra/Util.java

@@ -551,13 +551,15 @@ public class Util
     public static void spinAssertEquals(Object expected, Supplier<Object> s, int timeoutInSeconds)
     {
         long start = System.currentTimeMillis();
+        Object lastValue = null;
         while (System.currentTimeMillis() < start + (1000 * timeoutInSeconds))
         {
-            if (s.get().equals(expected))
+            lastValue = s.get();
+            if (lastValue.equals(expected))
                 break;
             Thread.yield();
         }
-        assertEquals(expected, s.get());
+        assertEquals(expected, lastValue);
     }
 
     public static void joinThread(Thread thread) throws InterruptedException

test/unit/org/apache/cassandra/db/fullquerylog/FullQueryLoggerTest.java

@@ -0,0 +1,601 @@
+/*
+ * 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.fullquerylog;
+
+
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.After;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.netty.buffer.Unpooled;
+import net.openhft.chronicle.queue.ChronicleQueue;
+import net.openhft.chronicle.queue.ChronicleQueueBuilder;
+import net.openhft.chronicle.queue.ExcerptTailer;
+import net.openhft.chronicle.queue.RollCycles;
+import net.openhft.chronicle.wire.ValueIn;
+import net.openhft.chronicle.wire.WireOut;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.db.fullquerylog.FullQueryLogger.WeighableMarshallableQuery;
+import org.apache.cassandra.db.fullquerylog.FullQueryLogger.WeighableMarshallableBatch;
+import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.transport.Server;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.binlog.BinLogTest;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class FullQueryLoggerTest
+{
+    private static Path tempDir;
+
+    @BeforeClass
+    public static void beforeClass() throws Exception
+    {
+        tempDir = BinLogTest.tempDir();
+    }
+
+    @After
+    public void tearDown()
+    {
+        FullQueryLogger.instance.reset(tempDir.toString());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testConfigureNullPath() throws Exception
+    {
+        FullQueryLogger.instance.configure(null, "", true, 1, 1);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testConfigureNullRollCycle() throws Exception
+    {
+        FullQueryLogger.instance.configure(BinLogTest.tempDir(), null, true, 1, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testConfigureInvalidRollCycle() throws Exception
+    {
+        FullQueryLogger.instance.configure(BinLogTest.tempDir(), "foobar", true, 1, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testConfigureInvalidMaxQueueWeight() throws Exception
+    {
+        FullQueryLogger.instance.configure(BinLogTest.tempDir(), "DAILY", true, 0, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testConfigureInvalidMaxQueueLogSize() throws Exception
+    {
+        FullQueryLogger.instance.configure(BinLogTest.tempDir(), "DAILY", true, 1, 0);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testConfigureOverExistingFile() throws Exception
+    {
+        File f = File.createTempFile("foo", "bar");
+        f.deleteOnExit();
+        FullQueryLogger.instance.configure(f.toPath(), "TEST_SECONDLY", true, 1, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testCanRead() throws Exception
+    {
+        tempDir.toFile().setReadable(false);
+        try
+        {
+            configureFQL();
+        }
+        finally
+        {
+            tempDir.toFile().setReadable(true);
+        }
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testCanWrite() throws Exception
+    {
+        tempDir.toFile().setWritable(false);
+        try
+        {
+            configureFQL();
+        }
+        finally
+        {
+            tempDir.toFile().setWritable(true);
+        }
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testCanExecute() throws Exception
+    {
+        tempDir.toFile().setExecutable(false);
+        try
+        {
+            configureFQL();
+        }
+        finally
+        {
+            tempDir.toFile().setExecutable(true);
+        }
+    }
+
+    @Test
+    public void testResetWithoutConfigure() throws Exception
+    {
+        FullQueryLogger.instance.reset(tempDir.toString());
+        FullQueryLogger.instance.reset(tempDir.toString());
+    }
+
+    @Test
+    public void stopWithoutConfigure() throws Exception
+    {
+        FullQueryLogger.instance.stop();
+        FullQueryLogger.instance.stop();
+    }
+
+    /**
+     * Both the last used and supplied directory should get cleaned
+     */
+    @Test
+    public void testResetCleansPaths() throws Exception
+    {
+        configureFQL();
+        File tempA = File.createTempFile("foo", "bar", tempDir.toFile());
+        assertTrue(tempA.exists());
+        File tempB = File.createTempFile("foo", "bar", BinLogTest.tempDir().toFile());
+        FullQueryLogger.instance.reset(tempB.getParent());
+        assertFalse(tempA.exists());
+        assertFalse(tempB.exists());
+    }
+
+    /**
+     * The last used and configured directory are the same and it shouldn't be an issue
+     */
+    @Test
+    public void testResetSamePath() throws Exception
+    {
+        configureFQL();
+        File tempA = File.createTempFile("foo", "bar", tempDir.toFile());
+        assertTrue(tempA.exists());
+        FullQueryLogger.instance.reset(tempA.getParent());
+        assertFalse(tempA.exists());
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void testDoubleConfigure() throws Exception
+    {
+        configureFQL();
+        configureFQL();
+    }
+
+    @Test
+    public void testCleansDirectory() throws Exception
+    {
+        assertTrue(new File(tempDir.toFile(), "foobar").createNewFile());
+        configureFQL();
+        assertEquals(tempDir.toFile().listFiles().length, 1);
+        assertEquals("directory-listing.cq4t", tempDir.toFile().listFiles()[0].getName());
+    }
+
+    @Test
+    public void testEnabledReset() throws Exception
+    {
+        assertFalse(FullQueryLogger.instance.enabled());
+        configureFQL();
+        assertTrue(FullQueryLogger.instance.enabled());
+        FullQueryLogger.instance.reset(tempDir.toString());
+        assertFalse(FullQueryLogger.instance.enabled());
+    }
+
+    @Test
+    public void testEnabledStop() throws Exception
+    {
+        assertFalse(FullQueryLogger.instance.enabled());
+        configureFQL();
+        assertTrue(FullQueryLogger.instance.enabled());
+        FullQueryLogger.instance.stop();
+        assertFalse(FullQueryLogger.instance.enabled());
+    }
+
+    /**
+     * Test that a thread will block if the FQL is over weight, and unblock once the backup is cleared
+     */
+    @Test
+    public void testBlocking() throws Exception
+    {
+        configureFQL();
+        //Prevent the bin log thread from making progress, causing the task queue to block
+        Semaphore blockBinLog = new Semaphore(0);
+        try
+        {
+            //Find out when the bin log thread has been blocked, necessary to not run into batch task drain behavior
+            Semaphore binLogBlocked = new Semaphore(0);
+            FullQueryLogger.instance.binLog.put(new WeighableMarshallableQuery("foo1", QueryOptions.DEFAULT, 1)
+            {
+
+                public void writeMarshallable(WireOut wire)
+                {
+                    //Notify that the bin log is blocking now
+                    binLogBlocked.release();
+                    try
+                    {
+                        //Block the bin log thread so the task queue can be filled
+                        blockBinLog.acquire();
+                    }
+                    catch (InterruptedException e)
+                    {
+                        e.printStackTrace();
+                    }
+                    super.writeMarshallable(wire);
+                }
+
+                public void release()
+                {
+                    super.release();
+                }
+            });
+
+            //Wait for the bin log thread to block so it can't batch drain tasks
+            Util.spinAssertEquals(true, binLogBlocked::tryAcquire, 60);
+
+            //Now fill the task queue
+            logQuery("foo2");
+
+            //Start a thread to block waiting on the bin log queue
+            Thread t = new Thread(() ->
+                                  {
+                                      logQuery("foo3");
+                                      //Should be able to log another query without an issue
+                                      logQuery("foo4");
+                                  });
+            t.start();
+            Thread.sleep(500);
+            //If thread state is terminated then the thread started, finished, and didn't block on the full task queue
+            assertTrue(t.getState() != Thread.State.TERMINATED);
+        }
+        finally
+        {
+            //Unblock the binlog thread
+            blockBinLog.release();
+        }
+        Util.spinAssertEquals(true, () -> checkForQueries(Arrays.asList("foo1", "foo2", "foo3", "foo4")), 60);
+    }
+
+    private boolean checkForQueries(List<String> queries)
+    {
+        try (ChronicleQueue queue = ChronicleQueueBuilder.single(tempDir.toFile()).rollCycle(RollCycles.TEST_SECONDLY).build())
+        {
+            ExcerptTailer tailer = queue.createTailer();
+            List<String> expectedQueries = new LinkedList<>(queries);
+            while (!expectedQueries.isEmpty())
+            {
+                if (!tailer.readDocument(wire -> {
+                    assertEquals(expectedQueries.get(0), wire.read("query").text());
+                    expectedQueries.remove(0);
+                }))
+                {
+                    return false;
+                }
+            }
+            assertFalse(tailer.readDocument(wire -> {}));
+            return true;
+        }
+    }
+
+    @Test
+    public void testNonBlocking() throws Exception
+    {
+        FullQueryLogger.instance.configure(tempDir, "TEST_SECONDLY", false, 1, 1024 * 1024 * 256);
+        //Prevent the bin log thread from making progress, causing the task queue to refuse tasks
+        Semaphore blockBinLog = new Semaphore(0);
+        try
+        {
+            //Find out when the bin log thread has been blocked, necessary to not run into batch task drain behavior
+            Semaphore binLogBlocked = new Semaphore(0);
+            FullQueryLogger.instance.binLog.put(new WeighableMarshallableQuery("foo1", QueryOptions.DEFAULT, 1)
+            {
+
+                public void writeMarshallable(WireOut wire)
+                {
+                    //Notify that the bin log is blocking now
+                    binLogBlocked.release();
+                    try
+                    {
+                        //Block the bin log thread so the task queue can be filled
+                        blockBinLog.acquire();
+                    }
+                    catch (InterruptedException e)
+                    {
+                        e.printStackTrace();
+                    }
+                    super.writeMarshallable(wire);
+                }
+
+                public void release()
+                {
+                    super.release();
+                }
+            });
+
+            //Wait for the bin log thread to block so it can't batch drain tasks
+            Util.spinAssertEquals(true, binLogBlocked::tryAcquire, 60);
+
+            //Now fill the task queue
+            logQuery("foo2");
+
+            //This sample should get dropped AKA released without being written
+            AtomicInteger releasedCount = new AtomicInteger(0);
+            AtomicInteger writtenCount = new AtomicInteger(0);
+            FullQueryLogger.instance.logRecord(new WeighableMarshallableQuery("foo3", QueryOptions.DEFAULT, 1) {
+                public void writeMarshallable(WireOut wire)
+                {
+                    writtenCount.incrementAndGet();
+                    super.writeMarshallable(wire);
+                }
+
+                public void release()
+                {
+                    releasedCount.incrementAndGet();
+                    super.release();
+                }
+            }, FullQueryLogger.instance.binLog);
+
+            Util.spinAssertEquals(1, releasedCount::get, 60);
+            assertEquals(0, writtenCount.get());
+        }
+        finally
+        {
+            blockBinLog.release();
+        }
+        //Wait for tasks to drain so there should be space in the queue
+        Util.spinAssertEquals(true, () -> checkForQueries(Arrays.asList("foo1", "foo2")), 60);
+        //Should be able to log again
+        logQuery("foo4");
+        Util.spinAssertEquals(true, () -> checkForQueries(Arrays.asList("foo1", "foo2", "foo4")), 60);
+    }
+
+    @Test
+    public void testRoundTripQuery() throws Exception
+    {
+        configureFQL();
+        logQuery("foo");
+        Util.spinAssertEquals(true, () -> checkForQueries(Arrays.asList("foo")), 60);
+        try (ChronicleQueue queue = ChronicleQueueBuilder.single(tempDir.toFile()).rollCycle(RollCycles.TEST_SECONDLY).build())
+        {
+            ExcerptTailer tailer = queue.createTailer();
+            assertTrue(tailer.readDocument(wire -> {
+                assertEquals("single", wire.read("type").text());
+                ProtocolVersion protocolVersion = ProtocolVersion.decode(wire.read("protocol-version").int32());
+                assertEquals(ProtocolVersion.CURRENT, protocolVersion);
+                QueryOptions queryOptions = QueryOptions.codec.decode(Unpooled.wrappedBuffer(wire.read("query-options").bytes()), protocolVersion);
+                compareQueryOptions(QueryOptions.DEFAULT, queryOptions);
+                assertEquals(1L, wire.read("query-time").int64());
+                assertEquals("foo", wire.read("query").text());
+            }));
+        }
+    }
+
+    @Test
+    public void testRoundTripBatch() throws Exception
+    {
+        configureFQL();
+        FullQueryLogger.instance.logBatch("UNLOGGED", Arrays.asList("foo1", "foo2"), Arrays.asList(Arrays.asList(ByteBuffer.allocate(1) , ByteBuffer.allocateDirect(2)), Arrays.asList()), QueryOptions.DEFAULT, 1);
+        Util.spinAssertEquals(true, () ->
+        {
+            try (ChronicleQueue queue = ChronicleQueueBuilder.single(tempDir.toFile()).rollCycle(RollCycles.TEST_SECONDLY).build())
+            {
+                return queue.createTailer().readingDocument().isPresent();
+            }
+        }, 60);
+        try (ChronicleQueue queue = ChronicleQueueBuilder.single(tempDir.toFile()).rollCycle(RollCycles.TEST_SECONDLY).build())
+        {
+            ExcerptTailer tailer = queue.createTailer();
+            assertTrue(tailer.readDocument(wire -> {
+                assertEquals("batch", wire.read("type").text());
+                ProtocolVersion protocolVersion = ProtocolVersion.decode(wire.read("protocol-version").int32());
+                assertEquals(ProtocolVersion.CURRENT, protocolVersion);
+                QueryOptions queryOptions = QueryOptions.codec.decode(Unpooled.wrappedBuffer(wire.read("query-options").bytes()), protocolVersion);
+                assertEquals(1L, wire.read("query-time").int64());
+                compareQueryOptions(QueryOptions.DEFAULT, queryOptions);
+                assertEquals("UNLOGGED", wire.read("batch-type").text());
+                ValueIn in = wire.read("queries");
+                assertEquals(2, in.int32());
+                assertEquals("foo1", in.text());
+                assertEquals("foo2", in.text());
+                in = wire.read("values");
+                assertEquals(2, in.int32());
+                assertEquals(2, in.int32());
+                assertTrue(Arrays.equals(new byte[1], in.bytes()));
+                assertTrue(Arrays.equals(new byte[2], in.bytes()));
+                assertEquals(0, in.int32());
+            }));
+        }
+    }
+
+    @Test
+    public void testQueryWeight()
+    {
+        //Empty query should have some weight
+        WeighableMarshallableQuery query = new WeighableMarshallableQuery("", QueryOptions.DEFAULT, 1);
+        assertTrue(query.weight() >= 95);
+
+        StringBuilder sb = new StringBuilder();
+        for (int ii = 0; ii < 1024 * 1024; ii++)
+        {
+            sb.append('a');
+        }
+        query = new WeighableMarshallableQuery(sb.toString(), QueryOptions.DEFAULT, 1);
+
+        //A large query should be reflected in the size, * 2 since characters are still two bytes
+        assertTrue(query.weight() > ObjectSizes.measureDeep(sb.toString()));
+
+        //Large query options should be reflected
+        QueryOptions largeOptions = QueryOptions.forInternalCalls(Arrays.asList(ByteBuffer.allocate(1024 * 1024)));
+        query = new WeighableMarshallableQuery("", largeOptions, 1);
+        assertTrue(query.weight() > 1024 * 1024);
+        System.out.printf("weight %d%n", query.weight());
+    }
+
+    @Test
+    public void testBatchWeight()
+    {
+        //An empty batch should have weight
+        WeighableMarshallableBatch batch = new WeighableMarshallableBatch("", new ArrayList<>(), new ArrayList<>(), QueryOptions.DEFAULT, 1);
+        assertTrue(batch.weight() >= 183);
+
+        StringBuilder sb = new StringBuilder();
+        for (int ii = 0; ii < 1024 * 1024; ii++)
+        {
+            sb.append('a');
+        }
+
+        //The weight of the type string should be reflected
+        batch = new WeighableMarshallableBatch(sb.toString(), new ArrayList<>(), new ArrayList<>(), QueryOptions.DEFAULT, 1);
+        assertTrue(batch.weight() > ObjectSizes.measureDeep(sb.toString()));
+
+        //The weight of the list containing queries should be reflected
+        List<String> bigList = new ArrayList(100000);
+        for (int ii = 0; ii < 100000; ii++)
+        {
+            bigList.add("");
+        }
+        batch = new WeighableMarshallableBatch("", bigList, new ArrayList<>(), QueryOptions.DEFAULT, 1);
+        assertTrue(batch.weight() > ObjectSizes.measureDeep(bigList));
+
+        //The size of the query should be reflected
+        bigList = new ArrayList(1);
+        bigList.add(sb.toString());
+        batch = new WeighableMarshallableBatch("", bigList, new ArrayList<>(), QueryOptions.DEFAULT, 1);
+        assertTrue(batch.weight() > ObjectSizes.measureDeep(bigList));
+
+        bigList = null;
+        //The size of the list of values should be reflected
+        List<List<ByteBuffer>> bigValues = new ArrayList<>(100000);
+        for (int ii = 0; ii < 100000; ii++)
+        {
+            bigValues.add(new ArrayList<>(0));
+        }
+        bigValues.get(0).add(ByteBuffer.allocate(1024 * 1024 * 5));
+        batch = new WeighableMarshallableBatch("", new ArrayList<>(),  bigValues, QueryOptions.DEFAULT, 1);
+        assertTrue(batch.weight() > ObjectSizes.measureDeep(bigValues));
+
+        //As should the size of the values
+        QueryOptions largeOptions = QueryOptions.forInternalCalls(Arrays.asList(ByteBuffer.allocate(1024 * 1024)));
+        batch = new WeighableMarshallableBatch("", new ArrayList<>(), new ArrayList<>(), largeOptions, 1);
+        assertTrue(batch.weight() > 1024 * 1024);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testLogBatchNullType() throws Exception
+    {
+        FullQueryLogger.instance.logBatch(null, new ArrayList<>(), new ArrayList<>(), QueryOptions.DEFAULT, 1);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testLogBatchNullQueries() throws Exception
+    {
+        FullQueryLogger.instance.logBatch("", null, new ArrayList<>(), QueryOptions.DEFAULT, 1);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testLogBatchNullQueriesQuery() throws Exception
+    {
+        configureFQL();
+        FullQueryLogger.instance.logBatch("", Arrays.asList((String)null), new ArrayList<>(), QueryOptions.DEFAULT, 1);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testLogBatchNullValues() throws Exception
+    {
+        FullQueryLogger.instance.logBatch("", new ArrayList<>(), null, QueryOptions.DEFAULT, 1);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testLogBatchNullValuesValue() throws Exception
+    {
+        FullQueryLogger.instance.logBatch("", new ArrayList<>(), Arrays.asList((List<ByteBuffer>)null), null, 1);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testLogBatchNullQueryOptions() throws Exception
+    {
+        FullQueryLogger.instance.logBatch("", new ArrayList<>(), new ArrayList<>(), null, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testLogBatchNegativeTime() throws Exception
+    {
+        FullQueryLogger.instance.logBatch("", new ArrayList<>(), new ArrayList<>(), QueryOptions.DEFAULT, -1);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testLogQueryNullQuery() throws Exception
+    {
+        FullQueryLogger.instance.logQuery(null, QueryOptions.DEFAULT, 1);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testLogQueryNullQueryOptions() throws Exception
+    {
+        FullQueryLogger.instance.logQuery("", null, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testLogQueryNegativeTime() throws Exception
+    {
+        FullQueryLogger.instance.logQuery("", QueryOptions.DEFAULT, -1);
+    }
+
+    private static void compareQueryOptions(QueryOptions a, QueryOptions b)
+    {
+        assertEquals(a.getClass(), b.getClass());
+        assertEquals(a.getProtocolVersion(), b.getProtocolVersion());
+        assertEquals(a.getPageSize(), b.getPageSize());
+        assertEquals(a.getConsistency(), b.getConsistency());
+        assertEquals(a.getPagingState(), b.getPagingState());
+        assertEquals(a.getValues(), b.getValues());
+        assertEquals(a.getSerialConsistency(), b.getSerialConsistency());
+    }
+
+    private void configureFQL() throws Exception
+    {
+        FullQueryLogger.instance.configure(tempDir, "TEST_SECONDLY", true, 1, 1024 * 1024 * 256);
+    }
+
+    private void logQuery(String query)
+    {
+        FullQueryLogger.instance.logQuery(query, QueryOptions.DEFAULT, 1);
+    }
+
+}
+

test/unit/org/apache/cassandra/utils/binlog/BinLogTest.java

@@ -0,0 +1,449 @@
+/*
+ * 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.utils.binlog;
+
+import java.io.File;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import net.openhft.chronicle.queue.ChronicleQueue;
+import net.openhft.chronicle.queue.ChronicleQueueBuilder;
+import net.openhft.chronicle.queue.ExcerptTailer;
+import net.openhft.chronicle.queue.RollCycles;
+import net.openhft.chronicle.wire.WireOut;
+import org.apache.cassandra.Util;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class BinLogTest
+{
+    public static Path tempDir() throws Exception
+    {
+        File f = File.createTempFile("foo", "bar");
+        f.delete();
+        f.mkdir();
+        return Paths.get(f.getPath());
+    }
+
+    private static final String testString = "ry@nlikestheyankees";
+    private static final String testString2 = testString + "1";
+
+    private BinLog binLog;
+    private Path path;
+
+    @Before
+    public void setUp() throws Exception
+    {
+        path = tempDir();
+        binLog = new BinLog(path, RollCycles.TEST_SECONDLY, 10, 1024 * 1024 * 128);
+        binLog.start();
+    }
+
+    @After
+    public void tearDown() throws Exception
+    {
+        if (binLog != null)
+        {
+            binLog.stop();
+        }
+        for (File f : path.toFile().listFiles())
+        {
+            f.delete();
+        }
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testConstructorNullPath() throws Exception
+    {
+        new BinLog(null, RollCycles.TEST_SECONDLY, 1, 1);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testConstructorNullRollCycle() throws Exception
+    {
+        new BinLog(tempDir(), null, 1, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testConstructorZeroWeight() throws Exception
+    {
+        new BinLog(tempDir(), RollCycles.TEST_SECONDLY, 0, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testConstructorLogSize() throws Exception
+    {
+        new BinLog(tempDir(), RollCycles.TEST_SECONDLY, 1, 0);
+    }
+
+    /**
+     * Check that we can start and stop the bin log and that it releases resources held by any subsequent appended
+     * records
+     */
+    @Test
+    public void testBinLogStartStop() throws Exception
+    {
+        Semaphore blockBinLog = new Semaphore(1);
+        AtomicInteger releaseCount = new AtomicInteger();
+        binLog.put(new BinLog.ReleaseableWriteMarshallable()
+        {
+            protected void release()
+            {
+                releaseCount.incrementAndGet();
+            }
+
+            public void writeMarshallable(WireOut wire)
+            {
+                try
+                {
+                    blockBinLog.acquire();
+                }
+                catch (InterruptedException e)
+                {
+                    throw new RuntimeException(e);
+                }
+            }
+        });
+        binLog.put(new BinLog.ReleaseableWriteMarshallable()
+        {
+
+            public void writeMarshallable(WireOut wire)
+            {
+
+            }
+
+            protected void release()
+            {
+                releaseCount.incrementAndGet();
+            }
+        });
+        Thread.sleep(1000);
+        assertEquals(2, releaseCount.get());
+        Thread t = new Thread(() -> {
+            try
+            {
+                binLog.stop();
+            }
+            catch (InterruptedException e)
+            {
+                throw new AssertionError(e);
+            }
+        });
+        t.start();
+        t.join(60 * 1000);
+        assertEquals(t.getState(), Thread.State.TERMINATED);
+
+        Util.spinAssertEquals(2, releaseCount::get, 60);
+        Util.spinAssertEquals(Thread.State.TERMINATED, binLog.binLogThread::getState, 60);
+     }
+
+    /**
+     * Check that the finalizer releases any stragglers in the queue
+     */
+    @Test
+    public void testBinLogFinalizer() throws Exception
+    {
+        binLog.stop();
+        Semaphore released = new Semaphore(0);
+        binLog.sampleQueue.put(new BinLog.ReleaseableWriteMarshallable()
+        {
+            protected void release()
+            {
+                released.release();
+            }
+
+            public void writeMarshallable(WireOut wire)
+            {
+
+            }
+        });
+        binLog = null;
+
+        for (int ii = 0; ii < 30; ii++)
+        {
+            System.gc();
+            System.runFinalization();
+            Thread.sleep(100);
+            if (released.tryAcquire())
+                return;
+        }
+        fail("Finalizer never released resources");
+    }
+
+    /**
+     * Test that put blocks and unblocks and creates records
+     */
+    @Test
+    public void testPut() throws Exception
+    {
+        binLog.put(record(testString));
+        binLog.put(record(testString2));
+
+        Util.spinAssertEquals(2, () -> readBinLogRecords(path).size(), 60);
+        List<String> records = readBinLogRecords(path);
+        assertEquals(testString, records.get(0));
+        assertEquals(testString2, records.get(1));
+
+
+        //Prevent the bin log thread from making progress
+        Semaphore blockBinLog = new Semaphore(0);
+        //Get notified when the bin log thread has blocked and definitely won't batch drain tasks
+        Semaphore binLogBlocked = new Semaphore(0);
+        try
+        {
+            binLog.put(new BinLog.ReleaseableWriteMarshallable()
+            {
+                protected void release()
+                {
+                }
+
+                public void writeMarshallable(WireOut wire)
+                {
+                    //Notify the bing log thread is about to block
+                    binLogBlocked.release();
+                    try
+                    {
+                        //Block the bin log thread so it doesn't process more tasks
+                        blockBinLog.acquire();
+                    }
+                    catch (InterruptedException e)
+                    {
+                        throw new RuntimeException(e);
+                    }
+                }
+            });
+
+            //Wait for the bin log thread to block so it doesn't batch drain
+            Util.spinAssertEquals(true, binLogBlocked::tryAcquire, 60);
+
+            //Now fill the queue up to capacity and it shouldn't block
+            for (int ii = 0; ii < 10; ii++)
+            {
+                binLog.put(record(testString));
+            }
+
+            //Thread to block on the full queue
+            Thread t = new Thread(() ->
+                                  {
+                                      try
+                                      {
+                                          binLog.put(record(testString));
+                                          //Should be able to do it again after unblocking
+                                          binLog.put(record(testString));
+                                      }
+                                      catch (InterruptedException e)
+                                      {
+                                          throw new AssertionError(e);
+                                      }
+                                  });
+            t.start();
+            Thread.sleep(500);
+            //If the thread is not terminated then it is probably blocked on the queue
+            assertTrue(t.getState() != Thread.State.TERMINATED);
+        }
+        finally
+        {
+            blockBinLog.release();
+        }
+
+        //Expect all the records to eventually be there including one from the blocked thread
+        Util.spinAssertEquals(15, () -> readBinLogRecords(path).size(), 60);
+    }
+
+    @Test
+    public void testOffer() throws Exception
+    {
+        assertTrue(binLog.offer(record(testString)));
+        assertTrue(binLog.offer(record(testString2)));
+
+        Util.spinAssertEquals(2, () -> readBinLogRecords(path).size(), 60);
+        List<String> records = readBinLogRecords(path);
+        assertEquals(testString, records.get(0));
+        assertEquals(testString2, records.get(1));
+
+        //Prevent the bin log thread from making progress
+        Semaphore blockBinLog = new Semaphore(0);
+        //Get notified when the bin log thread has blocked and definitely won't batch drain tasks
+        Semaphore binLogBlocked = new Semaphore(0);
+        try
+        {
+            assertTrue(binLog.offer(new BinLog.ReleaseableWriteMarshallable()
+            {
+                protected void release()
+                {
+                }
+
+                public void writeMarshallable(WireOut wire)
+                {
+                    //Notify the bing log thread is about to block
+                    binLogBlocked.release();
+                    try
+                    {
+                        //Block the bin log thread so it doesn't process more tasks
+                        blockBinLog.acquire();
+                    }
+                    catch (InterruptedException e)
+                    {
+                        throw new RuntimeException(e);
+                    }
+                }
+            }));
+
+            //Wait for the bin log thread to block so it doesn't batch drain
+            Util.spinAssertEquals(true, binLogBlocked::tryAcquire, 60);
+
+            //Now fill the queue up to capacity and it should always accept
+            for (int ii = 0; ii < 10; ii++)
+            {
+                assertTrue(binLog.offer(record(testString)));
+            }
+
+            //it shoudl reject this record since it is full
+            assertFalse(binLog.offer(record(testString)));
+        }
+        finally
+        {
+            blockBinLog.release();
+        }
+        Util.spinAssertEquals(13, () -> readBinLogRecords(path).size(), 60);
+        assertTrue(binLog.offer(record(testString)));
+        Util.spinAssertEquals(14, () -> readBinLogRecords(path).size(), 60);
+    }
+
+    /**
+     * Set a very small segment size so on rolling the segments are always deleted
+     */
+    @Test
+    public void testCleanupOnOversize() throws Exception
+    {
+        tearDown();
+        binLog = new BinLog(path, RollCycles.TEST_SECONDLY, 10000, 1);
+        binLog.start();
+        for (int ii = 0; ii < 5; ii++)
+        {
+            binLog.put(record(String.valueOf(ii)));
+            Thread.sleep(1001);
+        }
+        List<String> records = readBinLogRecords(path);
+        System.out.println("Records found are " + records);
+        assertTrue(records.size() < 5);
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void testNoReuse() throws Exception
+    {
+        binLog.stop();
+        binLog.start();
+    }
+
+    @Test
+    public void testOfferAfterStop() throws Exception
+    {
+        binLog.stop();
+        assertFalse(binLog.offer(record(testString)));
+    }
+
+    @Test
+    public void testPutAfterStop() throws Exception
+    {
+        binLog.stop();
+        binLog.put(record(testString));
+        assertEquals(null, binLog.sampleQueue.poll());
+    }
+
+    /**
+     * Test for a bug where files were deleted but the space was not reclaimed when tracking so
+     * all log segemnts were incorrectly deleted when rolled.
+     */
+    @Test
+    public void testTrucationReleasesLogSpace() throws Exception
+    {
+        StringBuilder sb = new StringBuilder();
+        for (int ii = 0; ii < 1024 * 1024 * 2; ii++)
+        {
+            sb.append('a');
+        }
+
+        String queryString = sb.toString();
+
+        //This should fill up the log so when it rolls in the future it will always delete the rolled segment;
+        for (int ii = 0; ii < 129; ii++)
+        {
+            binLog.put(record(queryString));
+        }
+
+        for (int ii = 0; ii < 2; ii++)
+        {
+            Thread.sleep(2000);
+            binLog.put(record(queryString));
+        }
+
+        Util.spinAssertEquals(2, () -> readBinLogRecords(path).size(), 60);
+    }
+
+    static BinLog.ReleaseableWriteMarshallable record(String text)
+    {
+        return new BinLog.ReleaseableWriteMarshallable()
+        {
+            protected void release()
+            {
+                //Do nothing
+            }
+
+            public void writeMarshallable(WireOut wire)
+            {
+                wire.write("text").text(text);
+            }
+        };
+    }
+
+    List<String> readBinLogRecords(Path path)
+    {
+        List<String> records = new ArrayList<String>();
+        try (ChronicleQueue queue = ChronicleQueueBuilder.single(path.toFile()).rollCycle(RollCycles.TEST_SECONDLY).build())
+        {
+            ExcerptTailer tailer = queue.createTailer();
+            while (true)
+            {
+                if (!tailer.readDocument(wire ->
+                                         {
+                                             records.add(wire.read("text").text());
+                                         }))
+                {
+                    return records;
+                }
+            }
+        }
+        catch (Exception e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+}

test/unit/org/apache/cassandra/utils/concurrent/WeightedQueueTest.java

@@ -0,0 +1,656 @@
+/*
+ * 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.utils.concurrent;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class WeightedQueueTest
+{
+    private static WeightedQueue<Object> queue()
+    {
+        return new WeightedQueue<>(10);
+    }
+
+    private WeightedQueue<Object> queue;
+
+    @Before
+    public void setUp()
+    {
+        queue = queue();
+    }
+
+    private static WeightedQueue.Weighable weighable(int weight)
+    {
+        return () -> weight;
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testAddUnsupported() throws Exception
+    {
+        queue.add(new Object());
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testRemoveUnsupported() throws Exception
+    {
+        queue.remove();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testElementUnsupported() throws Exception
+    {
+        queue.element();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testPeekUnsupported() throws Exception
+    {
+        queue.peek();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testRemainingCapacityUnsupported() throws Exception
+    {
+        queue.remainingCapacity();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testRemoveElementUnsupported() throws Exception
+    {
+        queue.remove(null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testContainsAllUnsupported() throws Exception
+    {
+        queue.containsAll(null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testAddAllUnsupported() throws Exception
+    {
+        queue.addAll(null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testRemoveAllUnsupported() throws Exception
+    {
+        queue.removeAll(null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testRetainAllUnsupported() throws Exception
+    {
+        queue.retainAll(null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testClearUnsupported() throws Exception
+    {
+        queue.clear();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testSizeUnsupported() throws Exception
+    {
+        queue.size();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testIsEmptyUnsupported() throws Exception
+    {
+        queue.isEmpty();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testContainsUnsupported() throws Exception
+    {
+        queue.contains(null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testIteratorUnsupported() throws Exception
+    {
+        queue.iterator();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testToArrayUnsupported() throws Exception
+    {
+        queue.toArray();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testToArray2Unsupported() throws Exception
+    {
+        queue.toArray( new Object[] {});
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testDrainToUnsupported() throws Exception
+    {
+        queue.drainTo(new ArrayList<>());
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testTimedPollUnsupported() throws Exception
+    {
+        queue.poll(1, TimeUnit.MICROSECONDS);
+    }
+
+    @Test
+    public void testDrainToWithLimit() throws Exception
+    {
+        queue.offer(new Object());
+        queue.offer(new Object());
+        queue.offer(new Object());
+        ArrayList<Object> list = new ArrayList<>();
+        queue.drainTo(list, 1);
+        assertEquals(1, list.size());
+        list.clear();
+        queue.drainTo(list, 10);
+        assertEquals(2, list.size());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void offerNullThrows() throws Exception
+    {
+        queue.offer(null);
+    }
+
+    /**
+     * This also tests that natural weight (weighable interface) is respected
+     */
+    @Test
+    public void offerFullFails() throws Exception
+    {
+        assertTrue(queue.offer(weighable(10)));
+        assertFalse(queue.offer(weighable(1)));
+    }
+
+    /**
+     * Validate permits aren't leaked and return values are correct
+     */
+    @Test
+    public void testOfferWrappedQueueRefuses() throws Exception
+    {
+        queue = new WeightedQueue<>(10, new BadQueue(true), WeightedQueue.NATURAL_WEIGHER);
+        assertEquals(10, queue.availableWeight.availablePermits());
+        assertFalse(queue.offer(new Object()));
+        assertEquals(10, queue.availableWeight.availablePermits());
+    }
+
+    /**
+     * Validate permits aren't leaked and return values are correct
+     */
+    @Test
+    public void testOfferWrappedQueueThrows() throws Exception
+    {
+        queue = new WeightedQueue<>(10, new BadQueue(false), WeightedQueue.NATURAL_WEIGHER);
+        assertEquals(10, queue.availableWeight.availablePermits());
+        try
+        {
+            assertFalse(queue.offer(new Object()));
+            fail();
+        }
+        catch (UnsupportedOperationException e)
+        {
+            //expected and desired
+        }
+        assertEquals(10, queue.availableWeight.availablePermits());
+    }
+
+    /**
+     * If not weighable and not custom weigher the default weight is 1
+     */
+    @Test
+    public void defaultWeightRespected() throws Exception
+    {
+        for (int ii = 0; ii < 10; ii++)
+        {
+            assertTrue(queue.offer(new Object()));
+        }
+        assertFalse(queue.offer(new Object()));
+    }
+
+    @Test
+    public void testCustomWeigher() throws Exception
+    {
+        queue = new WeightedQueue<>(10, new LinkedBlockingQueue<>(), weighable -> 10 );
+        assertTrue(queue.offer(new Object()));
+        assertFalse(queue.offer(new Object()));
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testCustomQueue() throws Exception
+    {
+        new WeightedQueue<>(10, new BadQueue(false), WeightedQueue.NATURAL_WEIGHER).offer(new Object());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void timedOfferNullValueThrows() throws Exception
+    {
+        queue.offer(null, 1, TimeUnit.SECONDS);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void timedOfferNullTimeThrows() throws Exception
+    {
+        queue.offer(null, 1, null);
+    }
+
+    /**
+     * This is how it seems to be handled in java.util.concurrent, it's the same as just try
+     */
+    @Test
+    public void timedOfferNegativeTimeIgnored() throws Exception
+    {
+        queue.offer(weighable(10));
+        queue.offer(new Object(), -1, TimeUnit.SECONDS);
+    }
+
+    /**
+     * This also tests that natural weight (weighable interface) is respected
+     */
+    @Test
+    public void timedOfferFullFails() throws Exception
+    {
+        assertTrue(queue.offer(weighable(10), 1, TimeUnit.MICROSECONDS));
+        assertFalse(queue.offer(weighable(1), 1, TimeUnit.MICROSECONDS));
+    }
+
+    @Test
+    public void timedOfferEventuallySucceeds() throws Exception
+    {
+        assertTrue(queue.offer(weighable(10), 1, TimeUnit.MICROSECONDS));
+        Thread t = new Thread(() ->
+          {
+              try
+              {
+                  queue.offer(weighable(1), 1, TimeUnit.DAYS);
+              }
+              catch (InterruptedException e)
+              {
+                  e.printStackTrace();
+              }
+          });
+        t.start();
+        Thread.sleep(100);
+        assertTrue(t.getState() != Thread.State.TERMINATED);
+        queue.poll();
+        t.join(60000);
+        assertEquals(t.getState(), Thread.State.TERMINATED);
+    }
+
+    /**
+     * Validate permits aren't leaked and return values are correct
+     */
+    @Test
+    public void testTimedOfferWrappedQueueRefuses() throws Exception
+    {
+        queue = new WeightedQueue<>(10, new BadQueue(true), WeightedQueue.NATURAL_WEIGHER);
+        assertEquals(10, queue.availableWeight.availablePermits());
+        assertFalse(queue.offer(new Object(), 1, TimeUnit.MICROSECONDS));
+        assertEquals(10, queue.availableWeight.availablePermits());
+    }
+
+    /**
+     * Validate permits aren't leaked and return values are correct
+     */
+    @Test
+    public void testTimedOfferWrappedQueueThrows() throws Exception
+    {
+        queue = new WeightedQueue<>(10, new BadQueue(false), WeightedQueue.NATURAL_WEIGHER);
+        assertEquals(10, queue.availableWeight.availablePermits());
+        try
+        {
+            assertFalse(queue.offer(new Object(), 1, TimeUnit.MICROSECONDS));
+            fail();
+        }
+        catch (UnsupportedOperationException e)
+        {
+            //expected and desired
+        }
+        assertEquals(10, queue.availableWeight.availablePermits());
+    }
+
+
+    @Test
+    public void testPoll() throws Exception
+    {
+        assertEquals(10, queue.availableWeight.availablePermits());
+        assertNull(queue.poll());
+        assertEquals(10, queue.availableWeight.availablePermits());
+        Object o = new Object();
+        assertTrue(queue.offer(o));
+        assertEquals(9, queue.availableWeight.availablePermits());
+        WeightedQueue.Weighable weighable = weighable(9);
+        assertTrue(queue.offer(weighable));
+        assertEquals(0, queue.availableWeight.availablePermits());
+        assertEquals(o, queue.poll());
+        assertEquals(1, queue.availableWeight.availablePermits());
+        assertEquals(weighable, queue.poll());
+        assertEquals(10, queue.availableWeight.availablePermits());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testPutNullThrows() throws Exception
+    {
+        queue.put(null);
+    }
+
+    @Test
+    public void testPutFullBlocks() throws Exception
+    {
+        WeightedQueue.Weighable weighable = weighable(10);
+        assertEquals(10, queue.availableWeight.availablePermits());
+        queue.put(weighable);
+        assertEquals(0, queue.availableWeight.availablePermits());
+        Object o = new Object();
+        Thread t = new Thread(() -> {
+            try
+            {
+                queue.put(o);
+            } catch (InterruptedException e)
+            {
+                e.printStackTrace();
+            }
+        });
+        t.start();
+        Thread.sleep(100);
+        assertTrue(t.getState() != Thread.State.TERMINATED);
+        assertEquals(0, queue.availableWeight.availablePermits());
+        assertEquals(weighable, queue.poll());
+        assertTrue(queue.availableWeight.availablePermits() > 0);
+        t.join();
+        assertEquals(o, queue.poll());
+        assertEquals(10, queue.availableWeight.availablePermits());
+    }
+
+    @Test
+    public void testPutWrappedQueueThrows() throws Exception
+    {
+        queue = new WeightedQueue<>(10, new BadQueue(false), WeightedQueue.NATURAL_WEIGHER);
+        assertEquals(10, queue.availableWeight.availablePermits());
+        try
+        {
+            queue.put(new Object());
+            fail();
+        }
+        catch (UnsupportedOperationException e)
+        {
+            //expected and desired
+        }
+        assertEquals(10, queue.availableWeight.availablePermits());
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testTryAcquireWeightIllegalWeight()
+    {
+        queue.tryAcquireWeight(weighable(-1));
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testAcquireWeightIllegalWeight() throws Exception
+    {
+        queue.acquireWeight(weighable(-1), 1, TimeUnit.DAYS);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testReleaseWeightIllegalWeight()
+    {
+        queue.releaseWeight(weighable(-1));
+    }
+
+    @Test
+    public void testTake() throws Exception
+    {
+        Thread t = new Thread(() -> {
+            try
+            {
+                queue.take();
+            }
+            catch (InterruptedException e)
+            {
+                e.printStackTrace();
+            }
+        });
+        t.start();
+        Thread.sleep(500);
+        assertTrue(t.getState() != Thread.State.TERMINATED);
+        assertEquals(10, queue.availableWeight.availablePermits());
+        queue.offer(new Object());
+        t.join(60 * 1000);
+        assertEquals(t.getState(), Thread.State.TERMINATED);
+        assertEquals(10, queue.availableWeight.availablePermits());
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testConstructorLTZeroWeightThrows() throws Exception
+    {
+        new WeightedQueue(0);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testConstructor2LTZeroWeightThrows() throws Exception
+    {
+        new WeightedQueue(0, new LinkedBlockingQueue<>(), WeightedQueue.NATURAL_WEIGHER);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testConstructorNullQueueThrows() throws Exception
+    {
+        new WeightedQueue(1, null, WeightedQueue.NATURAL_WEIGHER);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testConstructorNullWeigherThrows() throws Exception
+    {
+        new WeightedQueue(1, new LinkedBlockingQueue<>(), null);
+    }
+
+    /**
+     * A blocking queue that throws or refuses on every method
+     */
+    private static class BadQueue implements BlockingQueue<Object>
+    {
+        /**
+         * Refuse instead of throwing for some methods that have a boolean return value
+         */
+        private boolean refuse = false;
+
+        private BadQueue(boolean refuse)
+        {
+            this.refuse = refuse;
+        }
+
+        @Override
+        public boolean add(Object o)
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean offer(Object o)
+        {
+            if (refuse)
+            {
+                return false;
+            }
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Object remove()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Object poll()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Object element()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Object peek()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public void put(Object o) throws InterruptedException
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean offer(Object o, long timeout, TimeUnit unit) throws InterruptedException
+        {
+            if (refuse)
+            {
+                return false;
+            }
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Object take() throws InterruptedException
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Object poll(long timeout, TimeUnit unit) throws InterruptedException
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public int remainingCapacity()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean remove(Object o)
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean addAll(Collection c)
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public void clear()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean retainAll(Collection c)
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean removeAll(Collection c)
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean containsAll(Collection c)
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public int size()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean isEmpty()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean contains(Object o)
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Iterator iterator()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Object[] toArray()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Object[] toArray(Object[] a)
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public int drainTo(Collection c)
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public int drainTo(Collection c, int maxElements)
+        {
+            throw new UnsupportedOperationException();
+        }
+    };
+
+}