Merge pull request #193 from dpkp/test_apache_binary_release

Use Kafka Binaries for Integration Tests (Issue #176)

18 files changed, 475 insertions, 62 deletions

diff --git a/.gitignore b/.gitignore
index cc1d06f..8cf9c4e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,3 +5,6 @@ build
 dist
 MANIFEST
 env
+servers/*/kafka-bin
+.coverage
+.noseids
diff --git a/.gitmodules b/.gitmodules
index b40620c..e69de29 100644
--- a/.gitmodules
+++ b/.gitmodules
@@ -1,6 +0,0 @@
-[submodule "servers/0.8.0/kafka-src"]
-	path = servers/0.8.0/kafka-src
-	url = https://github.com/apache/kafka.git
-[submodule "servers/0.8.1/kafka-src"]
-	path = servers/0.8.1/kafka-src
-	url = https://github.com/apache/kafka.git
diff --git a/.travis.yml b/.travis.yml
index bd5f63a..22b03a3 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -5,8 +5,13 @@ python:
     - 2.7
     - pypy
 
+env:
+    - 
+    - KAFKA_VERSION=0.8.0
+    - KAFKA_VERSION=0.8.1
+    - KAFKA_VERSION=0.8.1.1
+
 before_install:
-    - git submodule update --init --recursive
     - sudo apt-get install libsnappy-dev
     - ./build_integration.sh
 
@@ -19,5 +24,3 @@ install:
 
 script:
     - tox -e `./travis_selector.sh $TRAVIS_PYTHON_VERSION`
-    - KAFKA_VERSION=0.8.0 tox -e `./travis_selector.sh $TRAVIS_PYTHON_VERSION`
-    - KAFKA_VERSION=0.8.1 tox -e `./travis_selector.sh $TRAVIS_PYTHON_VERSION`
diff --git a/README.md b/README.md
index 8e99124..c056318 100644
--- a/README.md
+++ b/README.md
@@ -190,32 +190,33 @@ pip install python-snappy
 tox
 ```
 
+## Run a single unit test
+```shell
+tox -e py27 -- -v --with-id 102
+```
+
 ## Run the integration tests
 
 The integration tests will actually start up real local Zookeeper
 instance and Kafka brokers, and send messages in using the client.
 
-Note that you may want to add this to your global gitignore:
+First, get the kafka binaries for integration testing:
 ```shell
-.gradle/
-clients/build/
-contrib/build/
-contrib/hadoop-consumer/build/
-contrib/hadoop-producer/build/
-core/build/
-core/data/
-examples/build/
-perf/build/
+./build_integration.sh
 ```
-
-First, check out and the Kafka source:
+By default, the build_integration.sh script will download binary
+distributions for all supported kafka versions.
+To test against the latest source build, set KAFKA_VERSION=trunk
+and optionally set SCALA_VERSION (defaults to 2.8.0, but 2.10.1 is recommended)
 ```shell
-git submodule update --init
-./build_integration.sh
+SCALA_VERSION=2.10.1 KAFKA_VERSION=trunk ./build_integration.sh
 ```
 
 Then run the tests against supported Kafka versions:
 ```shell
 KAFKA_VERSION=0.8.0 tox
 KAFKA_VERSION=0.8.1 tox
+KAFKA_VERSION=0.8.1.1 tox
+KAFKA_VERSION=trunk tox
 ```
+
diff --git a/build_integration.sh b/build_integration.sh
index da87f34..bb46e54 100755
--- a/build_integration.sh
+++ b/build_integration.sh
@@ -1,5 +1,61 @@
 #!/bin/bash
 
-git submodule update --init
-(cd servers/0.8.0/kafka-src && ./sbt update package assembly-package-dependency)
-(cd servers/0.8.1/kafka-src && ./gradlew jar)
+# Versions available for testing via binary distributions
+OFFICIAL_RELEASES="0.8.0 0.8.1 0.8.1.1"
+
+# Useful configuration vars, with sensible defaults
+if [ -z "$SCALA_VERSION" ]; then
+  SCALA_VERSION=2.8.0
+fi
+
+# On travis CI, empty KAFKA_VERSION means skip integration tests
+# so we dont try to get binaries 
+# Otherwise it means test all official releases, so we get all of them!
+if [ -z "$KAFKA_VERSION" -a -z "$TRAVIS" ]; then
+  KAFKA_VERSION=$OFFICIAL_RELEASES
+fi
+
+# By default look for binary releases at archive.apache.org
+if [ -z "$DIST_BASE_URL" ]; then
+  DIST_BASE_URL="https://archive.apache.org/dist/kafka/"
+fi
+
+# When testing against source builds, use this git repo
+if [ -z "$KAFKA_SRC_GIT" ]; then
+  KAFKA_SRC_GIT="https://github.com/apache/kafka.git"
+fi
+
+pushd servers
+  mkdir -p dist
+  pushd dist
+    for kafka in $KAFKA_VERSION; do
+      if [ "$kafka" == "trunk" ]; then
+        if [ ! -d "$kafka" ]; then
+          git clone $KAFKA_SRC_GIT $kafka
+        fi
+        pushd $kafka
+          git pull
+          ./gradlew -PscalaVersion=$SCALA_VERSION -Pversion=$kafka releaseTarGz -x signArchives
+        popd
+        # Not sure how to construct the .tgz name accurately, so use a wildcard (ugh)
+        tar xzvf $kafka/core/build/distributions/kafka_*.tgz -C ../$kafka/
+        rm $kafka/core/build/distributions/kafka_*.tgz
+        mv ../$kafka/kafka_* ../$kafka/kafka-bin
+      else
+        echo "-------------------------------------"
+        echo "Checking kafka binaries for ${kafka}"
+        echo
+        wget -N https://archive.apache.org/dist/kafka/$kafka/kafka_${SCALA_VERSION}-${kafka}.tgz || wget -N https://archive.apache.org/dist/kafka/$kafka/kafka_${SCALA_VERSION}-${kafka}.tar.gz
+        echo
+        if [ ! -d "../$kafka/kafka-bin" ]; then
+          echo "Extracting kafka binaries for ${kafka}"
+          tar xzvf kafka_${SCALA_VERSION}-${kafka}.t* -C ../$kafka/
+          mv ../$kafka/kafka_${SCALA_VERSION}-${kafka} ../$kafka/kafka-bin
+        else
+          echo "$kafka/kafka-bin directory already exists -- skipping tgz extraction"
+        fi
+      fi
+      echo
+    done
+  popd
+popd
diff --git a/servers/0.8.0/kafka-src b/servers/0.8.0/kafka-src
deleted file mode 160000
-Subproject 15bb3961d9171c1c54c4c840a554ce2c7616816
diff --git a/servers/0.8.1.1/resources/kafka.properties b/servers/0.8.1.1/resources/kafka.properties
new file mode 100644
index 0000000..a638f39
--- /dev/null
+++ b/servers/0.8.1.1/resources/kafka.properties
@@ -0,0 +1,118 @@
+# 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.
+# see kafka.server.KafkaConfig for additional details and defaults
+
+############################# Server Basics #############################
+
+# The id of the broker. This must be set to a unique integer for each broker.
+broker.id={broker_id}
+
+############################# Socket Server Settings #############################
+
+# The port the socket server listens on
+port={port}
+
+# Hostname the broker will bind to. If not set, the server will bind to all interfaces
+host.name={host}
+
+# Hostname the broker will advertise to producers and consumers. If not set, it uses the
+# value for "host.name" if configured.  Otherwise, it will use the value returned from
+# java.net.InetAddress.getCanonicalHostName().
+#advertised.host.name=<hostname routable by clients>
+
+# The port to publish to ZooKeeper for clients to use. If this is not set,
+# it will publish the same port that the broker binds to.
+#advertised.port=<port accessible by clients>
+
+# The number of threads handling network requests
+num.network.threads=2
+ 
+# The number of threads doing disk I/O
+num.io.threads=8
+
+# The send buffer (SO_SNDBUF) used by the socket server
+socket.send.buffer.bytes=1048576
+
+# The receive buffer (SO_RCVBUF) used by the socket server
+socket.receive.buffer.bytes=1048576
+
+# The maximum size of a request that the socket server will accept (protection against OOM)
+socket.request.max.bytes=104857600
+
+
+############################# Log Basics #############################
+
+# A comma seperated list of directories under which to store log files
+log.dirs={tmp_dir}/data
+
+# The default number of log partitions per topic. More partitions allow greater
+# parallelism for consumption, but this will also result in more files across
+# the brokers.
+num.partitions={partitions}
+default.replication.factor={replicas}
+
+############################# Log Flush Policy #############################
+
+# Messages are immediately written to the filesystem but by default we only fsync() to sync
+# the OS cache lazily. The following configurations control the flush of data to disk. 
+# There are a few important trade-offs here:
+#    1. Durability: Unflushed data may be lost if you are not using replication.
+#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
+#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks. 
+# The settings below allow one to configure the flush policy to flush data after a period of time or
+# every N messages (or both). This can be done globally and overridden on a per-topic basis.
+
+# The number of messages to accept before forcing a flush of data to disk
+#log.flush.interval.messages=10000
+
+# The maximum amount of time a message can sit in a log before we force a flush
+#log.flush.interval.ms=1000
+
+############################# Log Retention Policy #############################
+
+# The following configurations control the disposal of log segments. The policy can
+# be set to delete segments after a period of time, or after a given size has accumulated.
+# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
+# from the end of the log.
+
+# The minimum age of a log file to be eligible for deletion
+log.retention.hours=168
+
+# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
+# segments don't drop below log.retention.bytes.
+#log.retention.bytes=1073741824
+
+# The maximum size of a log segment file. When this size is reached a new log segment will be created.
+log.segment.bytes=536870912
+
+# The interval at which log segments are checked to see if they can be deleted according 
+# to the retention policies
+log.retention.check.interval.ms=60000
+
+# By default the log cleaner is disabled and the log retention policy will default to just delete segments after their retention expires.
+# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
+log.cleaner.enable=false
+
+############################# Zookeeper #############################
+
+# Zookeeper connection string (see zookeeper docs for details).
+# This is a comma separated host:port pairs, each corresponding to a zk
+# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
+# You can also append an optional chroot string to the urls to specify the
+# root directory for all kafka znodes.
+zookeeper.connect={zk_host}:{zk_port}/{zk_chroot}
+
+# Timeout in ms for connecting to zookeeper
+zookeeper.connection.timeout.ms=1000000
diff --git a/servers/0.8.1.1/resources/log4j.properties b/servers/0.8.1.1/resources/log4j.properties
new file mode 100644
index 0000000..f863b3b
--- /dev/null
+++ b/servers/0.8.1.1/resources/log4j.properties
@@ -0,0 +1,24 @@
+# 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.
+
+log4j.rootLogger=INFO, stdout
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+log4j.logger.kafka=DEBUG, stdout
+log4j.logger.org.I0Itec.zkclient.ZkClient=INFO, stdout
+log4j.logger.org.apache.zookeeper=INFO, stdout
diff --git a/servers/0.8.1.1/resources/zookeeper.properties b/servers/0.8.1.1/resources/zookeeper.properties
new file mode 100644
index 0000000..e3fd097
--- /dev/null
+++ b/servers/0.8.1.1/resources/zookeeper.properties
@@ -0,0 +1,21 @@
+# 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.
+# the directory where the snapshot is stored.
+dataDir={tmp_dir}
+# the port at which the clients will connect
+clientPort={port}
+clientPortAddress={host}
+# disable the per-ip limit on the number of connections since this is a non-production config
+maxClientCnxns=0
diff --git a/servers/0.8.1/kafka-src b/servers/0.8.1/kafka-src
deleted file mode 160000
-Subproject 150d0a70cbe2b1f980e9565a4fa59b0420d1c0a
diff --git a/servers/trunk/resources/kafka.properties b/servers/trunk/resources/kafka.properties
new file mode 100644
index 0000000..a638f39
--- /dev/null
+++ b/servers/trunk/resources/kafka.properties
@@ -0,0 +1,118 @@
+# 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.
+# see kafka.server.KafkaConfig for additional details and defaults
+
+############################# Server Basics #############################
+
+# The id of the broker. This must be set to a unique integer for each broker.
+broker.id={broker_id}
+
+############################# Socket Server Settings #############################
+
+# The port the socket server listens on
+port={port}
+
+# Hostname the broker will bind to. If not set, the server will bind to all interfaces
+host.name={host}
+
+# Hostname the broker will advertise to producers and consumers. If not set, it uses the
+# value for "host.name" if configured.  Otherwise, it will use the value returned from
+# java.net.InetAddress.getCanonicalHostName().
+#advertised.host.name=<hostname routable by clients>
+
+# The port to publish to ZooKeeper for clients to use. If this is not set,
+# it will publish the same port that the broker binds to.
+#advertised.port=<port accessible by clients>
+
+# The number of threads handling network requests
+num.network.threads=2
+ 
+# The number of threads doing disk I/O
+num.io.threads=8
+
+# The send buffer (SO_SNDBUF) used by the socket server
+socket.send.buffer.bytes=1048576
+
+# The receive buffer (SO_RCVBUF) used by the socket server
+socket.receive.buffer.bytes=1048576
+
+# The maximum size of a request that the socket server will accept (protection against OOM)
+socket.request.max.bytes=104857600
+
+
+############################# Log Basics #############################
+
+# A comma seperated list of directories under which to store log files
+log.dirs={tmp_dir}/data
+
+# The default number of log partitions per topic. More partitions allow greater
+# parallelism for consumption, but this will also result in more files across
+# the brokers.
+num.partitions={partitions}
+default.replication.factor={replicas}
+
+############################# Log Flush Policy #############################
+
+# Messages are immediately written to the filesystem but by default we only fsync() to sync
+# the OS cache lazily. The following configurations control the flush of data to disk. 
+# There are a few important trade-offs here:
+#    1. Durability: Unflushed data may be lost if you are not using replication.
+#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
+#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks. 
+# The settings below allow one to configure the flush policy to flush data after a period of time or
+# every N messages (or both). This can be done globally and overridden on a per-topic basis.
+
+# The number of messages to accept before forcing a flush of data to disk
+#log.flush.interval.messages=10000
+
+# The maximum amount of time a message can sit in a log before we force a flush
+#log.flush.interval.ms=1000
+
+############################# Log Retention Policy #############################
+
+# The following configurations control the disposal of log segments. The policy can
+# be set to delete segments after a period of time, or after a given size has accumulated.
+# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
+# from the end of the log.
+
+# The minimum age of a log file to be eligible for deletion
+log.retention.hours=168
+
+# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
+# segments don't drop below log.retention.bytes.
+#log.retention.bytes=1073741824
+
+# The maximum size of a log segment file. When this size is reached a new log segment will be created.
+log.segment.bytes=536870912
+
+# The interval at which log segments are checked to see if they can be deleted according 
+# to the retention policies
+log.retention.check.interval.ms=60000
+
+# By default the log cleaner is disabled and the log retention policy will default to just delete segments after their retention expires.
+# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
+log.cleaner.enable=false
+
+############################# Zookeeper #############################
+
+# Zookeeper connection string (see zookeeper docs for details).
+# This is a comma separated host:port pairs, each corresponding to a zk
+# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
+# You can also append an optional chroot string to the urls to specify the
+# root directory for all kafka znodes.
+zookeeper.connect={zk_host}:{zk_port}/{zk_chroot}
+
+# Timeout in ms for connecting to zookeeper
+zookeeper.connection.timeout.ms=1000000
diff --git a/servers/trunk/resources/log4j.properties b/servers/trunk/resources/log4j.properties
new file mode 100644
index 0000000..f863b3b
--- /dev/null
+++ b/servers/trunk/resources/log4j.properties
@@ -0,0 +1,24 @@
+# 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.
+
+log4j.rootLogger=INFO, stdout
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+log4j.logger.kafka=DEBUG, stdout
+log4j.logger.org.I0Itec.zkclient.ZkClient=INFO, stdout
+log4j.logger.org.apache.zookeeper=INFO, stdout
diff --git a/servers/trunk/resources/zookeeper.properties b/servers/trunk/resources/zookeeper.properties
new file mode 100644
index 0000000..e3fd097
--- /dev/null
+++ b/servers/trunk/resources/zookeeper.properties
@@ -0,0 +1,21 @@
+# 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.
+# the directory where the snapshot is stored.
+dataDir={tmp_dir}
+# the port at which the clients will connect
+clientPort={port}
+clientPortAddress={host}
+# disable the per-ip limit on the number of connections since this is a non-production config
+maxClientCnxns=0
diff --git a/test/fixtures.py b/test/fixtures.py
index df8cd42..adb0642 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -1,9 +1,11 @@
 import logging
 import glob
 import os
+import os.path
 import shutil
 import subprocess
 import tempfile
+import urllib2
 import uuid
 
 from urlparse import urlparse
@@ -14,42 +16,63 @@ class Fixture(object):
     kafka_version = os.environ.get('KAFKA_VERSION', '0.8.0')
     scala_version = os.environ.get("SCALA_VERSION", '2.8.0')
     project_root = os.environ.get('PROJECT_ROOT', os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-    kafka_root = os.environ.get("KAFKA_ROOT", os.path.join(project_root, 'servers', kafka_version, "kafka-src"))
+    kafka_root = os.environ.get("KAFKA_ROOT", os.path.join(project_root, 'servers', kafka_version, "kafka-bin"))
     ivy_root = os.environ.get('IVY_ROOT', os.path.expanduser("~/.ivy2/cache"))
 
     @classmethod
-    def test_resource(cls, filename):
-        return os.path.join(cls.project_root, "servers", cls.kafka_version, "resources", filename)
+    def download_official_distribution(cls,
+                                       kafka_version=None,
+                                       scala_version=None,
+                                       output_dir=None):
+        if not kafka_version:
+            kafka_version = cls.kafka_version
+        if not scala_version:
+            scala_version = cls.scala_version
+        if not output_dir:
+            output_dir = os.path.join(cls.project_root, 'servers', 'dist')
+
+        distfile = 'kafka_%s-%s' % (scala_version, kafka_version,)
+        url_base = 'https://archive.apache.org/dist/kafka/%s/' % (kafka_version,)
+        output_file = os.path.join(output_dir, distfile + '.tgz')
+
+        if os.path.isfile(output_file):
+            logging.info("Found file already on disk: %s" % output_file)
+            return output_file
+
+        # New tarballs are .tgz, older ones are sometimes .tar.gz
+        try:
+            url = url_base + distfile + '.tgz'
+            logging.info("Attempting to download %s" % (url,))
+            response = urllib2.urlopen(url)
+        except urllib2.HTTPError:
+            logging.exception("HTTP Error")
+            url = url_base + distfile + '.tar.gz'
+            logging.info("Attempting to download %s" % (url,))
+            response = urllib2.urlopen(url)
+
+        logging.info("Saving distribution file to %s" % (output_file,))
+        with open(os.path.join(output_dir, distfile + '.tgz'), 'w') as f:
+            f.write(response.read())
+
+        return output_file
 
     @classmethod
-    def test_classpath(cls):
-        # ./kafka-src/bin/kafka-run-class.sh is the authority.
-        jars = ["."]
-
-        # 0.8.0 build path, should contain the core jar and a deps jar
-        jars.extend(glob.glob(cls.kafka_root + "/core/target/scala-%s/*.jar" % cls.scala_version))
-
-        # 0.8.1 build path, should contain the core jar and several dep jars
-        jars.extend(glob.glob(cls.kafka_root + "/core/build/libs/*.jar"))
-        jars.extend(glob.glob(cls.kafka_root + "/core/build/dependant-libs-%s/*.jar" % cls.scala_version))
-
-        jars = filter(os.path.exists, map(os.path.abspath, jars))
-        return ":".join(jars)
+    def test_resource(cls, filename):
+        return os.path.join(cls.project_root, "servers", cls.kafka_version, "resources", filename)
 
     @classmethod
     def kafka_run_class_args(cls, *args):
-        # ./kafka-src/bin/kafka-run-class.sh is the authority.
-        result = ["java", "-Xmx512M", "-server"]
-        result.append("-Dlog4j.configuration=file:%s" % cls.test_resource("log4j.properties"))
-        result.append("-Dcom.sun.management.jmxremote")
-        result.append("-Dcom.sun.management.jmxremote.authenticate=false")
-        result.append("-Dcom.sun.management.jmxremote.ssl=false")
-        result.append("-cp")
-        result.append(cls.test_classpath())
+        result = [os.path.join(cls.kafka_root, 'bin', 'kafka-run-class.sh')]
         result.extend(args)
         return result
 
     @classmethod
+    def kafka_run_class_env(cls):
+        env = os.environ.copy()
+        env['KAFKA_LOG4J_OPTS'] = "-Dlog4j.configuration=file:%s" % cls.test_resource("log4j.properties")
+        return env
+
+    @classmethod
     def render_template(cls, source_file, target_file, binding):
         with open(source_file, "r") as handle:
             template = handle.read()
@@ -94,10 +117,11 @@ class ZookeeperFixture(Fixture):
         self.render_template(template, properties, vars(self))
 
         # Configure Zookeeper child process
-        self.child = SpawnedService(self.kafka_run_class_args(
+        self.child = SpawnedService(args=self.kafka_run_class_args(
             "org.apache.zookeeper.server.quorum.QuorumPeerMain",
-            properties
-        ))
+            properties),
+            env=self.kafka_run_class_env()
+        )
 
         # Party!
         self.out("Starting...")
@@ -175,9 +199,10 @@ class KafkaFixture(Fixture):
         self.render_template(template, properties, vars(self))
 
         # Configure Kafka child process
-        self.child = SpawnedService(self.kafka_run_class_args(
-            "kafka.Kafka", properties
-        ))
+        self.child = SpawnedService(args=self.kafka_run_class_args(
+            "kafka.Kafka", properties),
+            env=self.kafka_run_class_env()
+        )
 
         # Party!
         self.out("Creating Zookeeper chroot node...")
@@ -186,6 +211,7 @@ class KafkaFixture(Fixture):
                 "-server", "%s:%d" % (self.zk_host, self.zk_port),
                 "create", "/%s" % self.zk_chroot, "kafka-python"
             ),
+            env=self.kafka_run_class_env(),
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE)
 
diff --git a/test/service.py b/test/service.py
index 8872c82..df6c1ba 100644
--- a/test/service.py
+++ b/test/service.py
@@ -26,10 +26,11 @@ class ExternalService(object):
 
 
 class SpawnedService(threading.Thread):
-    def __init__(self, args=[]):
+    def __init__(self, args=[], env=None):
         threading.Thread.__init__(self)
 
         self.args = args
+        self.env = env
         self.captured_stdout = []
         self.captured_stderr = []
 
@@ -41,6 +42,7 @@ class SpawnedService(threading.Thread):
     def run_with_handles(self):
         self.child = subprocess.Popen(
             self.args,
+            env=self.env,
             bufsize=1,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE)
@@ -78,7 +80,7 @@ class SpawnedService(threading.Thread):
         for line in self.captured_stdout:
             logging.critical(line.rstrip())
 
-    def wait_for(self, pattern, timeout=10):
+    def wait_for(self, pattern, timeout=30):
         t1 = time.time()
         while True:
             t2 = time.time()
@@ -89,11 +91,13 @@ class SpawnedService(threading.Thread):
                     logging.exception("Received exception when killing child process")
                 self.dump_logs()
 
-                raise RuntimeError("Waiting for %r timed out" % pattern)
+                raise RuntimeError("Waiting for %r timed out after %d seconds" % (pattern, timeout))
 
             if re.search(pattern, '\n'.join(self.captured_stdout), re.IGNORECASE) is not None:
+                logging.info("Found pattern %r in %d seconds via stdout", pattern, (t2 - t1))
                 return
             if re.search(pattern, '\n'.join(self.captured_stderr), re.IGNORECASE) is not None:
+                logging.info("Found pattern %r in %d seconds via stderr", pattern, (t2 - t1))
                 return
             time.sleep(0.1)
 
diff --git a/test/test_client_integration.py b/test/test_client_integration.py
index 261d168..49c4b57 100644
--- a/test/test_client_integration.py
+++ b/test/test_client_integration.py
@@ -53,7 +53,7 @@ class TestKafkaClientIntegration(KafkaIntegrationTestCase):
     #   Offset Tests   #
     ####################
 
-    @kafka_versions("0.8.1")
+    @kafka_versions("0.8.1", "0.8.1.1")
     def test_commit_fetch_offsets(self):
         req = OffsetCommitRequest(self.topic, 0, 42, "metadata")
         (resp,) = self.client.send_offset_commit_request("group", [req])
diff --git a/test/test_consumer_integration.py b/test/test_consumer_integration.py
index a6589b3..cb32669 100644
--- a/test/test_consumer_integration.py
+++ b/test/test_consumer_integration.py
@@ -213,7 +213,7 @@ class TestConsumerIntegration(KafkaIntegrationTestCase):
 
         big_consumer.stop()
 
-    @kafka_versions("0.8.1")
+    @kafka_versions("0.8.1", "0.8.1.1")
     def test_offset_behavior__resuming_behavior(self):
         msgs1 = self.send_messages(0, range(0, 100))
         msgs2 = self.send_messages(1, range(100, 200))
diff --git a/tox.ini b/tox.ini
index 3c5fd17..09ec1e3 100644
--- a/tox.ini
+++ b/tox.ini
@@ -8,6 +8,6 @@ deps =
     mock
     python-snappy
 commands =
-    nosetests --with-coverage --cover-erase --cover-package kafka []
+    nosetests {posargs:-v --with-id --with-coverage --cover-erase --cover-package kafka}
 setenv =
     PROJECT_ROOT = {toxinidir}