KAFKA-5919: Adding checks on "version" field for tools using it

.gitignore

@@ -47,3 +47,7 @@ tests/venv
 docs/generated/
 
 .release-settings.json
+
+kafkatest.egg-info/
+systest/
+*.swp

README.md

@@ -4,9 +4,9 @@ See our [web site](http://kafka.apache.org) for details on the project.
 
 You need to have [Gradle](http://www.gradle.org/installation) and [Java](http://www.oracle.com/technetwork/java/javase/downloads/index.html) installed.
 
-Kafka requires Gradle 3.0 or higher.
+Kafka requires Gradle 4.5 or higher.
 
-Java 7 should be used for building in order to support both Java 7 and Java 8 at runtime.
+Java 8 should be used for building in order to support both Java 8 and Java 10 at runtime.
 
 ### First bootstrap and download the wrapper ###
     cd kafka_source_dir
@@ -85,8 +85,6 @@ You can pass either the major version (eg 2.11) or the full version (eg 2.11.12)
     ./gradlew -PscalaVersion=2.11 test
     ./gradlew -PscalaVersion=2.11 releaseTarGz
 
-Scala 2.12.x requires Java 8.
-
 ### Running a task for a specific project ###
 This is for `core`, `examples` and `clients`
 

TROGDOR.md

@@ -0,0 +1,154 @@
+Trogdor
+========================================
+Trogdor is a test framework for Apache Kafka.
+
+Trogdor can run benchmarks and other workloads.  Trogdor can also inject faults in order to stress test the system.
+
+Quickstart
+=========================================================
+First, we want to start a single-node Kafka cluster with a ZooKeeper and a broker.
+
+Running ZooKeeper:
+
+    > ./bin/zookeeper-server-start.sh ./config/zookeeper.properties &> /tmp/zookeeper.log &
+
+Running Kafka:
+
+    > ./bin/kafka-server-start.sh ./config/server.properties &> /tmp/kafka.log &
+
+Then, we want to run a Trogdor Agent, plus a Trogdor broker.
+
+To run the Trogdor Agent:
+
+    > ./bin/trogdor.sh agent -c ./config/trogdor.conf -n node0 &> /tmp/trogdor-agent.log &
+
+To run the Trogdor Coordinator:
+
+    > ./bin/trogdor.sh coordinator -c ./config/trogdor.conf -n node0 &> /tmp/trogdor-coordinator.log &
+
+Let's confirm that all of the daemons are running:
+
+    > jps
+    116212 Coordinator
+    115188 QuorumPeerMain
+    116571 Jps
+    115420 Kafka
+    115694 Agent
+
+Now, we can submit a test job to Trogdor.  Here's an example of a short bash script which makes it easier.
+
+    > ./tests/bin/trogdor-run-produce-bench.sh
+    [2018-04-12 10:32:04,055] DEBUG Sending POST with input {"id":"produce_bench_22137","spec":{"class":"org.apache.kafka.trogdor.workload.ProduceBenchSpec","startMs":0,"durationMs":10000000,"producerNode":"node0","bootstrapServers":"localhost:9092","targetMessagesPerSec":10,"maxMessages":100,"keyGenerator":{"type":"sequential","size":4,"startOffset":0},"valueGenerator":{"type":"constant","size":512,"value":"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="},"totalTopics":10,"activeTopics":5,"topicPrefix":"foo","replicationFactor":1,"classLoader":{},"numPartitions":1}} to http://localhost:8889/coordinator/task/create (org.apache.kafka.trogdor.coordinator.CoordinatorClient)
+    Created task.
+    $TASK_ID = produce_bench_20462
+
+To get the test results, we run --show-tasks:
+
+    ./tests/bin/trogdor.sh client --show-tasks localhost:8889
+    Got coordinator tasks: {
+      "tasks" : {
+        "produce_bench_20462" : {
+          "state" : "DONE",
+          "spec" : {
+            "class" : "org.apache.kafka.trogdor.workload.ProduceBenchSpec",
+            "startMs" : 0,
+            "durationMs" : 10000000,
+            "producerNode" : "node0",
+            "bootstrapServers" : "localhost:9092",
+            "targetMessagesPerSec" : 10,
+            "maxMessages" : 100,
+            "keyGenerator" : {
+              "type" : "sequential",
+              "size" : 4,
+              "startOffset" : 0
+            },
+            "valueGenerator" : {
+              "type" : "constant",
+              "size" : 512,
+              "value" : "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+            },
+            "totalTopics" : 10,
+            "activeTopics" : 5,
+            "topicPrefix" : "foo",
+            "replicationFactor" : 1,
+            "classLoader" : { },
+            "numPartitions" : 1
+          },
+          "startedMs" : 1523552769850,
+          "doneMs" : 1523552780878,
+          "cancelled" : false,
+          "status" : {
+            "totalSent" : 500,
+            "averageLatencyMs" : 4.972,
+            "p50LatencyMs" : 4,
+            "p95LatencyMs" : 6,
+            "p99LatencyMs" : 12
+          }
+        }
+      }
+    }
+
+Trogdor Architecture
+========================================
+Trogdor has a single coordinator process which manages multiple agent processes.  Each agent process is responsible for a single cluster node.
+
+The Trogdor coordinator manages tasks.  A task is anything we might want to do on a cluster, such as running a benchmark, injecting a fault, or running a workload.  In order to implement each task, the coordinator creates workers on one or more agent nodes.
+
+The Trogdor agent process implements the tasks.  For example, when running a workload, the agent process is the process which produces and consumes messages.
+
+Both the coordinator and the agent expose a REST interface that accepts objects serialized via JSON.  There is also a command-line program which makes it easy to send messages to either one without manually crafting the JSON message body.
+
+All Trogdor RPCs are idempotent except the shutdown requests.  Sending an idempotent RPC twice in a row has the same effect as sending the RPC once.
+
+Tasks
+========================================
+Tasks are described by specifications containing:
+
+* A "class" field describing the task type.  This contains a full Java class name.
+* A "startMs" field describing when the task should start.  This is given in terms of milliseconds since the UNIX epoch.
+* A "durationMs" field describing how long the task should last.  This is given in terms of milliseconds.
+* Other fields which are task-specific.
+
+The task specification is usually written as JSON.  For example, this task specification describes a network partition between nodes 1 and 2, and 3:
+
+    {
+        "class": "org.apache.kafka.trogdor.fault.NetworkPartitionFaultSpec",
+        "startMs": 1000,
+        "durationMs": 30000,
+        "partitions": [["node1", "node2"], ["node3"]]
+    }
+
+Tasks are submitted to the coordinator.  Once the coordinator determines that it is time for the task to start, it creates workers on agent processes.  The workers run until the task is done.
+
+Task specifications are immutable; they do not change after the task has been created.
+
+Tasks can be in several states:
+* PENDING, when task is waiting to execute,
+* RUNNING, when the task is running,
+* STOPPING, when the task is in the process of stopping,
+* DONE, when the task is done.
+
+Tasks that are DONE also have an error field which will be set if the task failed.
+
+Workloads
+========================================
+Trogdor can run several workloads.  Workloads perform operations on the cluster and measure their performance.  Workloads fail when the operations cannot be performed.
+
+### ProduceBench
+ProduceBench starts a Kafka producer on a single agent node, producing to several partitions.  The workload measures the average produce latency, as well as the median, 95th percentile, and 99th percentile latency.
+
+### RoundTripWorkload
+RoundTripWorkload tests both production and consumption.  The workload starts a Kafka producer and consumer on a single node.  The consumer will read back the messages that were produced by the producer.
+
+### ConsumeBench
+ConsumeBench starts a Kafka consumer on a single agent node.  The workload measures the average produce latency, as well as the median, 95th percentile, and 99th percentile latency.
+
+Faults
+========================================
+Trogdor can run several faults which deliberately break something in the cluster.
+
+### ProcessStopFault
+ProcessStopFault stops a process by sending it a SIGSTOP signal.  When the fault ends, the process is resumed with SIGCONT.
+
+### NetworkPartitionFault
+NetworkPartitionFault sets up an artificial network partition between one or more sets of nodes.  Currently, this is implemented using iptables.  The iptables rules are set up on the outbound traffic from the affected nodes.  Therefore, the affected nodes should still be reachable from outside the cluster.

Vagrantfile

@@ -40,7 +40,7 @@ ec2_keypair_file = nil
 
 ec2_region = "us-east-1"
 ec2_az = nil # Uses set by AWS
-ec2_ami = "ami-9eaa1cf6"
+ec2_ami = "ami-905730e8"
 ec2_instance_type = "m3.medium"
 ec2_user = "ubuntu"
 ec2_instance_name_prefix = "kafka-vagrant"
@@ -80,7 +80,7 @@ Vagrant.configure(VAGRANTFILE_API_VERSION) do |config|
       # share to a temporary location and the provisioning scripts symlink data
       # to the right location.
       override.cache.enable :generic, {
-        "oracle-jdk7" => { cache_dir: "/tmp/oracle-jdk7-installer-cache" },
+        "oracle-jdk8" => { cache_dir: "/tmp/oracle-jdk8-installer-cache" },
       }
     end
   end
@@ -141,7 +141,10 @@ Vagrant.configure(VAGRANTFILE_API_VERSION) do |config|
   def name_node(node, name, ec2_instance_name_prefix)
     node.vm.hostname = name
     node.vm.provider :aws do |aws|
-      aws.tags = { 'Name' => ec2_instance_name_prefix + "-" + Socket.gethostname + "-" + name }
+      aws.tags = {
+        'Name' => ec2_instance_name_prefix + "-" + Socket.gethostname + "-" + name,
+        'JenkinsBuildUrl' => ENV['BUILD_URL']
+      }
     end
   end
 

bin/kafka-dump-log.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+# 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.
+
+exec $(dirname $0)/kafka-run-class.sh kafka.tools.DumpLogSegments "$@"

bin/kafka-run-class.sh

@@ -69,28 +69,50 @@ do
   fi
 done
 
-for file in "$base_dir"/clients/build/libs/kafka-clients*.jar;
-do
-  if should_include_file "$file"; then
-    CLASSPATH="$CLASSPATH":"$file"
-  fi
-done
+if [ -z "$UPGRADE_KAFKA_STREAMS_TEST_VERSION" ]; then
+  clients_lib_dir=$(dirname $0)/../clients/build/libs
+  streams_lib_dir=$(dirname $0)/../streams/build/libs
+  rocksdb_lib_dir=$(dirname $0)/../streams/build/dependant-libs-${SCALA_VERSION}
+else
+  clients_lib_dir=/opt/kafka-$UPGRADE_KAFKA_STREAMS_TEST_VERSION/libs
+  streams_lib_dir=$clients_lib_dir
+  rocksdb_lib_dir=$streams_lib_dir
+fi
+
 
-for file in "$base_dir"/streams/build/libs/kafka-streams*.jar;
+for file in "$clients_lib_dir"/kafka-clients*.jar;
 do
   if should_include_file "$file"; then
     CLASSPATH="$CLASSPATH":"$file"
   fi
 done
 
-for file in "$base_dir"/streams/examples/build/libs/kafka-streams-examples*.jar;
+for file in "$streams_lib_dir"/kafka-streams*.jar;
 do
   if should_include_file "$file"; then
     CLASSPATH="$CLASSPATH":"$file"
   fi
 done
 
-for file in "$base_dir"/streams/build/dependant-libs-${SCALA_VERSION}/rocksdb*.jar;
+if [ -z "$UPGRADE_KAFKA_STREAMS_TEST_VERSION" ]; then
+  for file in "$base_dir"/streams/examples/build/libs/kafka-streams-examples*.jar;
+  do
+    if should_include_file "$file"; then
+      CLASSPATH="$CLASSPATH":"$file"
+    fi
+  done
+else
+  VERSION_NO_DOTS=`echo $UPGRADE_KAFKA_STREAMS_TEST_VERSION | sed 's/\.//g'`
+  SHORT_VERSION_NO_DOTS=${VERSION_NO_DOTS:0:((${#VERSION_NO_DOTS} - 1))} # remove last char, ie, bug-fix number
+  for file in "$base_dir"/streams/upgrade-system-tests-$SHORT_VERSION_NO_DOTS/build/libs/kafka-streams-upgrade-system-tests*.jar;
+  do
+    if should_include_file "$file"; then
+      CLASSPATH="$CLASSPATH":"$file"
+    fi
+  done
+fi
+
+for file in "$rocksdb_lib_dir"/rocksdb*.jar;
 do
   CLASSPATH="$CLASSPATH":"$file"
 done
@@ -137,7 +159,7 @@ done
 shopt -u nullglob
 
 if [ -z "$CLASSPATH" ] ; then
-  echo "Classpath is empty. Please build the project first e.g. by running './gradlew jar -Pscala_version=$SCALA_VERSION'"
+  echo "Classpath is empty. Please build the project first e.g. by running './gradlew jar -PscalaVersion=$SCALA_VERSION'"
   exit 1
 fi
 
@@ -216,6 +238,12 @@ if [ -z "$KAFKA_JVM_PERFORMANCE_OPTS" ]; then
   KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true"
 fi
 
+# version option
+for args in "$@" ; do
+  if [ "$args" = "--version" ]; then
+    exec $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "kafka.utils.VersionInfo"
+  fi
+done
 
 while [ $# -gt 0 ]; do
   COMMAND=$1
@@ -246,9 +274,16 @@ GC_FILE_SUFFIX='-gc.log'
 GC_LOG_FILE_NAME=''
 if [ "x$GC_LOG_ENABLED" = "xtrue" ]; then
   GC_LOG_FILE_NAME=$DAEMON_NAME$GC_FILE_SUFFIX
-  # the first segment of the version number, which is '1' for releases before Java 9
+
+  # The first segment of the version number, which is '1' for releases before Java 9
   # it then becomes '9', '10', ...
-  JAVA_MAJOR_VERSION=$($JAVA -version 2>&1 | sed -E -n 's/.* version "([^.-]*).*"/\1/p')
+  # Some examples of the first line of `java --version`:
+  # 8 -> java version "1.8.0_152"
+  # 9.0.4 -> java version "9.0.4"
+  # 10 -> java version "10" 2018-03-20
+  # 10.0.1 -> java version "10.0.1" 2018-04-17
+  # We need to match to the end of the line to prevent sed from printing the characters that do not match
+  JAVA_MAJOR_VERSION=$($JAVA -version 2>&1 | sed -E -n 's/.* version "([0-9]*).*$/\1/p')
   if [[ "$JAVA_MAJOR_VERSION" -ge "9" ]] ; then
     KAFKA_GC_LOG_OPTS="-Xlog:gc*:file=$LOG_DIR/$GC_LOG_FILE_NAME:time,tags:filecount=10,filesize=102400"
   else

bin/windows/kafka-consumer-offset-checker.bat → bin/windows/kafka-dump-log.bat

@@ -14,4 +14,4 @@ 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.
 
-"%~dp0kafka-run-class.bat" kafka.tools.ConsumerOffsetChecker %*
+"%~dp0kafka-run-class.bat" kafka.tools.DumpLogSegments %*