diff --git a/.gitignore b/.gitignore
index 60883492f3127..04f8feed0ad70 100644
--- a/.gitignore
+++ b/.gitignore
@@ -47,3 +47,7 @@ tests/venv
docs/generated/
.release-settings.json
+
+kafkatest.egg-info/
+systest/
+*.swp
diff --git a/README.md b/README.md
index 9371e48216986..18cb03e988048 100644
--- a/README.md
+++ b/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`
diff --git a/TROGDOR.md b/TROGDOR.md
new file mode 100644
index 0000000000000..b8c94d4bc7ce2
--- /dev/null
+++ b/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.
diff --git a/Vagrantfile b/Vagrantfile
index 2179a3c37ecb1..363607691fd0d 100644
--- a/Vagrantfile
+++ b/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
diff --git a/bin/kafka-dump-log.sh b/bin/kafka-dump-log.sh
new file mode 100755
index 0000000000000..a97ea7d3d9f8c
--- /dev/null
+++ b/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 "$@"
diff --git a/bin/kafka-run-class.sh b/bin/kafka-run-class.sh
index 6676d0d2156b7..0ea8b83194684 100755
--- a/bin/kafka-run-class.sh
+++ b/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
diff --git a/bin/windows/kafka-consumer-offset-checker.bat b/bin/windows/kafka-dump-log.bat
similarity index 92%
rename from bin/windows/kafka-consumer-offset-checker.bat
rename to bin/windows/kafka-dump-log.bat
index 2baa1b84061a0..3a1473dc61bc7 100644
--- a/bin/windows/kafka-consumer-offset-checker.bat
+++ b/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 %*
diff --git a/build.gradle b/build.gradle
index 5e4c35643c2a1..4f3fd770b78a3 100644
--- a/build.gradle
+++ b/build.gradle
@@ -26,9 +26,9 @@ buildscript {
// For Apache Rat plugin to ignore non-Git files
classpath "org.ajoberstar:grgit:1.9.3"
classpath 'com.github.ben-manes:gradle-versions-plugin:0.17.0'
- classpath 'org.scoverage:gradle-scoverage:2.1.0'
- classpath 'com.github.jengelman.gradle.plugins:shadow:2.0.2'
- classpath 'org.owasp:dependency-check-gradle:3.0.2'
+ classpath 'org.scoverage:gradle-scoverage:2.3.0'
+ classpath 'com.github.jengelman.gradle.plugins:shadow:2.0.4'
+ classpath 'org.owasp:dependency-check-gradle:3.1.2'
}
}
@@ -41,7 +41,6 @@ allprojects {
}
apply plugin: 'idea'
- apply plugin: "jacoco"
apply plugin: 'org.owasp.dependencycheck'
apply plugin: 'com.github.ben-manes.versions'
@@ -68,24 +67,19 @@ allprojects {
}
}
- if (JavaVersion.current().isJava8Compatible()) {
- tasks.withType(Javadoc) {
- // disable the crazy super-strict doclint tool in Java 8
- // noinspection SpellCheckingInspection
- options.addStringOption('Xdoclint:none', '-quiet')
- }
+ tasks.withType(Javadoc) {
+ // disable the crazy super-strict doclint tool in Java 8
+ // noinspection SpellCheckingInspection
+ options.addStringOption('Xdoclint:none', '-quiet')
}
}
ext {
- gradleVersion = "4.5.1"
+ gradleVersion = "4.7"
+ minJavaVersion = "8"
buildVersionFileName = "kafka-version.properties"
- maxPermSizeArgs = []
- if (!JavaVersion.current().isJava8Compatible())
- maxPermSizeArgs += '-XX:MaxPermSize=512m'
-
userMaxForks = project.hasProperty('maxParallelForks') ? maxParallelForks.toInteger() : null
skipSigning = project.hasProperty('skipSigning') && skipSigning.toBoolean()
@@ -121,11 +115,13 @@ if (new File('.git').exists()) {
'PULL_REQUEST_TEMPLATE.md',
'gradlew',
'gradlew.bat',
+ 'TROGDOR.md',
'**/README.md',
'**/id_rsa',
'**/id_rsa.pub',
'checkstyle/suppressions.xml',
- 'streams/quickstart/java/src/test/resources/projects/basic/goal.txt'
+ 'streams/quickstart/java/src/test/resources/projects/basic/goal.txt',
+ 'streams/streams-scala/logs/*'
])
}
}
@@ -144,20 +140,17 @@ subprojects {
if (!JavaVersion.current().isJava9Compatible())
apply plugin: 'findbugs'
- sourceCompatibility = 1.7
- targetCompatibility = 1.7
+ sourceCompatibility = minJavaVersion
+ targetCompatibility = minJavaVersion
compileJava {
options.encoding = 'UTF-8'
- options.compilerArgs << "-Xlint:deprecation"
- // -Xlint:unchecked is too buggy in Java 7, so we only enable for Java 8 or higher
- if (JavaVersion.current().isJava8Compatible())
- options.compilerArgs << "-Xlint:unchecked"
+ options.compilerArgs << "-Xlint:deprecation,unchecked"
// --release is the recommended way to select the target release, but it's only supported in Java 9 so we also
// set --source and --target via `sourceCompatibility` and `targetCompatibility`. If/when Gradle supports `--release`
// natively (https://github.com/gradle/gradle/issues/2510), we should switch to that.
if (JavaVersion.current().isJava9Compatible())
- options.compilerArgs << "--release" << "7"
+ options.compilerArgs << "--release" << minJavaVersion
}
uploadArchives {
@@ -201,7 +194,6 @@ subprojects {
minHeapSize = "256m"
maxHeapSize = "2048m"
- jvmArgs = maxPermSizeArgs
testLogging {
events = userTestLoggingEvents ?: testLoggingEvents
@@ -216,7 +208,6 @@ subprojects {
minHeapSize = "256m"
maxHeapSize = "2048m"
- jvmArgs = maxPermSizeArgs
testLogging {
events = userTestLoggingEvents ?: testLoggingEvents
@@ -235,7 +226,6 @@ subprojects {
minHeapSize = "256m"
maxHeapSize = "2048m"
- jvmArgs = maxPermSizeArgs
testLogging {
events = userTestLoggingEvents ?: testLoggingEvents
@@ -342,7 +332,7 @@ subprojects {
"-Xlint:private-shadow",
"-Xlint:stars-align",
"-Xlint:type-parameter-shadow",
- "-Xlint:unsound-match",
+ "-Xlint:unsound-match"
]
if (versions.baseScala != '2.11') {
@@ -354,15 +344,14 @@ subprojects {
configure(scalaCompileOptions.forkOptions) {
memoryMaximumSize = '1g'
- jvmArgs = ['-Xss2m'] + maxPermSizeArgs
+ jvmArgs = ['-Xss2m']
}
}
checkstyle {
configFile = new File(rootDir, "checkstyle/checkstyle.xml")
configProperties = [importControlFile: "$rootDir/checkstyle/import-control.xml"]
- // version 7.x requires Java 8
- toolVersion = '6.19'
+ toolVersion = '8.10'
}
test.dependsOn('checkstyleMain', 'checkstyleTest')
@@ -384,9 +373,10 @@ subprojects {
// Ignore core since its a scala project
if (it.path != ':core') {
+ apply plugin: "jacoco"
jacoco {
- toolVersion = "0.8.0"
+ toolVersion = "0.8.1"
}
// NOTE: Jacoco Gradle plugin does not support "offline instrumentation" this means that classes mocked by PowerMock
@@ -518,7 +508,7 @@ for ( sv in availableScalaVersions ) {
}
def connectPkgs = ['connect:api', 'connect:runtime', 'connect:transforms', 'connect:json', 'connect:file']
-def pkgs = ['clients', 'examples', 'log4j-appender', 'tools', 'streams', 'streams:test-utils', 'streams:examples'] + connectPkgs
+def pkgs = ['clients', 'examples', 'log4j-appender', 'tools', 'streams', 'streams:streams-scala', 'streams:test-utils', 'streams:examples'] + connectPkgs
/** Create one task per default Scala version */
def withDefScalaVersions(taskName) {
@@ -591,8 +581,7 @@ project(':core') {
scoverage libs.scoveragePlugin
scoverage libs.scoverageRuntime
}
-
- jacocoTestReport.enabled = false
+
scoverage {
reportDir = file("${rootProject.buildDir}/scoverage")
highlighting = false
@@ -634,6 +623,13 @@ project(':core') {
standardOutput = new File(generatedDocsDir, "protocol_errors.html").newOutputStream()
}
+ task genProtocolTypesDocs(type: JavaExec) {
+ classpath = sourceSets.main.runtimeClasspath
+ main = 'org.apache.kafka.common.protocol.types.Type'
+ if( !generatedDocsDir.exists() ) { generatedDocsDir.mkdirs() }
+ standardOutput = new File(generatedDocsDir, "protocol_types.html").newOutputStream()
+ }
+
task genProtocolApiKeyDocs(type: JavaExec) {
classpath = sourceSets.main.runtimeClasspath
main = 'org.apache.kafka.common.protocol.ApiKeys'
@@ -697,7 +693,7 @@ project(':core') {
standardOutput = new File(generatedDocsDir, "producer_metrics.html").newOutputStream()
}
- task siteDocsTar(dependsOn: ['genProtocolErrorDocs', 'genProtocolApiKeyDocs', 'genProtocolMessageDocs',
+ task siteDocsTar(dependsOn: ['genProtocolErrorDocs', 'genProtocolTypesDocs', 'genProtocolApiKeyDocs', 'genProtocolMessageDocs',
'genAdminClientConfigDocs', 'genProducerConfigDocs', 'genConsumerConfigDocs',
'genKafkaConfigDocs', 'genTopicConfigDocs',
':connect:runtime:genConnectConfigDocs', ':connect:runtime:genConnectTransformationDocs',
@@ -734,6 +730,8 @@ project(':core') {
from(project(':connect:file').configurations.runtime) { into("libs/") }
from(project(':streams').jar) { into("libs/") }
from(project(':streams').configurations.runtime) { into("libs/") }
+ from(project(':streams:streams-scala').jar) { into("libs/") }
+ from(project(':streams:streams-scala').configurations.runtime) { into("libs/") }
from(project(':streams:test-utils').jar) { into("libs/") }
from(project(':streams:test-utils').configurations.runtime) { into("libs/") }
from(project(':streams:examples').jar) { into("libs/") }
@@ -789,6 +787,7 @@ project(':clients') {
compile libs.lz4
compile libs.snappy
compile libs.slf4jApi
+ compile libs.jacksonDatabind // for SASL/OAUTHBEARER bearer token parsing
testCompile libs.bcpkix
testCompile libs.junit
@@ -859,6 +858,8 @@ project(':clients') {
include "**/org/apache/kafka/common/config/*"
include "**/org/apache/kafka/common/security/auth/*"
include "**/org/apache/kafka/server/policy/*"
+ include "**/org/apache/kafka/common/security/token/delegation/*"
+ include "**/org/apache/kafka/common/security/oauthbearer/*"
}
}
@@ -874,6 +875,9 @@ project(':tools') {
compile libs.jacksonJaxrsJsonProvider
compile libs.jerseyContainerServlet
+ compile libs.jerseyHk2
+ compile libs.jaxbApi // Jersey dependency that was available in the JDK before Java 9
+ compile libs.activation // Jersey dependency that was available in the JDK before Java 9
compile libs.jettyServer
compile libs.jettyServlet
compile libs.jettyServlets
@@ -918,13 +922,17 @@ project(':streams') {
compile libs.slf4jApi
compile libs.rocksDBJni
+ // testCompileOnly prevents streams from exporting a dependency on test-utils, which would cause a dependency cycle
+ testCompileOnly project(':streams:test-utils')
testCompile project(':clients').sourceSets.test.output
testCompile project(':core')
testCompile project(':core').sourceSets.test.output
+ testCompile libs.log4j
testCompile libs.junit
testCompile libs.easymock
testCompile libs.bcpkix
+ testRuntimeOnly project(':streams:test-utils')
testRuntime libs.slf4jlog4j
}
@@ -961,6 +969,45 @@ project(':streams') {
}
}
+project(':streams:streams-scala') {
+ println "Building project 'streams-scala' with Scala version ${versions.scala}"
+ apply plugin: 'scala'
+ archivesBaseName = "kafka-streams-scala_${versions.baseScala}"
+
+ dependencies {
+ compile project(':streams')
+
+ compile libs.scalaLibrary
+
+ testCompile project(':core')
+ testCompile project(':core').sourceSets.test.output
+ testCompile project(':streams').sourceSets.test.output
+ testCompile project(':clients').sourceSets.test.output
+
+ testCompile libs.junit
+ testCompile libs.scalatest
+
+ testRuntime libs.slf4jlog4j
+ }
+
+ javadoc {
+ include "**/org/apache/kafka/streams/scala/**"
+ }
+
+ tasks.create(name: "copyDependantLibs", type: Copy) {
+ from (configurations.runtime) {
+ exclude('kafka-streams*')
+ }
+ into "$buildDir/dependant-libs-${versions.scala}"
+ duplicatesStrategy 'exclude'
+ }
+
+ jar {
+ dependsOn 'copyDependantLibs'
+ }
+
+}
+
project(':streams:test-utils') {
archivesBaseName = "kafka-streams-test-utils"
@@ -1000,6 +1047,10 @@ project(':streams:examples') {
compile project(':streams')
compile project(':connect:json') // this dependency should be removed after we unify data API
compile libs.slf4jlog4j
+
+ testCompile project(':streams:test-utils')
+ testCompile project(':clients').sourceSets.test.output // for org.apache.kafka.test.IntegrationTest
+ testCompile libs.junit
}
javadoc {
@@ -1019,6 +1070,78 @@ project(':streams:examples') {
}
}
+project(':streams:upgrade-system-tests-0100') {
+ archivesBaseName = "kafka-streams-upgrade-system-tests-0100"
+
+ dependencies {
+ testCompile libs.kafkaStreams_0100
+ }
+
+ systemTestLibs {
+ dependsOn testJar
+ }
+}
+
+project(':streams:upgrade-system-tests-0101') {
+ archivesBaseName = "kafka-streams-upgrade-system-tests-0101"
+
+ dependencies {
+ testCompile libs.kafkaStreams_0101
+ }
+
+ systemTestLibs {
+ dependsOn testJar
+ }
+}
+
+project(':streams:upgrade-system-tests-0102') {
+ archivesBaseName = "kafka-streams-upgrade-system-tests-0102"
+
+ dependencies {
+ testCompile libs.kafkaStreams_0102
+ }
+
+ systemTestLibs {
+ dependsOn testJar
+ }
+}
+
+project(':streams:upgrade-system-tests-0110') {
+ archivesBaseName = "kafka-streams-upgrade-system-tests-0110"
+
+ dependencies {
+ testCompile libs.kafkaStreams_0110
+ }
+
+ systemTestLibs {
+ dependsOn testJar
+ }
+}
+
+project(':streams:upgrade-system-tests-10') {
+ archivesBaseName = "kafka-streams-upgrade-system-tests-10"
+
+ dependencies {
+ testCompile libs.kafkaStreams_10
+ }
+
+ systemTestLibs {
+ dependsOn testJar
+ }
+}
+
+project(':streams:upgrade-system-tests-11') {
+ archivesBaseName = "kafka-streams-upgrade-system-tests-11"
+
+ dependencies {
+ testCompile libs.kafkaStreams_11
+ }
+
+ systemTestLibs {
+ dependsOn testJar
+ }
+}
+
project(':jmh-benchmarks') {
apply plugin: 'com.github.johnrengelman.shadow'
@@ -1084,6 +1207,7 @@ project(':connect:api') {
dependencies {
compile project(':clients')
compile libs.slf4jApi
+ compile libs.jerseyContainerServlet
testCompile libs.junit
@@ -1195,14 +1319,19 @@ project(':connect:runtime') {
archivesBaseName = "connect-runtime"
dependencies {
+
compile project(':connect:api')
- compile project(":connect:transforms")
compile project(':clients')
compile project(':tools')
- compile libs.slf4jApi
+ compile project(':connect:json')
+ compile project(':connect:transforms')
+ compile libs.slf4jApi
compile libs.jacksonJaxrsJsonProvider
compile libs.jerseyContainerServlet
+ compile libs.jerseyHk2
+ compile libs.jaxbApi // Jersey dependency that was available in the JDK before Java 9
+ compile libs.activation // Jersey dependency that was available in the JDK before Java 9
compile libs.jettyServer
compile libs.jettyServlet
compile libs.jettyServlets
@@ -1216,7 +1345,6 @@ project(':connect:runtime') {
testCompile libs.powermockJunit4
testCompile libs.powermockEasymock
- testCompile project(":connect:json")
testCompile project(':clients').sourceSets.test.output
testRuntime libs.slf4jlog4j
@@ -1304,6 +1432,45 @@ project(':connect:file') {
}
}
+project(':connect:basic-auth-extension') {
+ archivesBaseName = "connect-basic-auth-extension"
+
+ dependencies {
+ compile project(':connect:api')
+ compile libs.slf4jApi
+
+ testCompile libs.bcpkix
+ testCompile libs.easymock
+ testCompile libs.junit
+ testCompile libs.powermockJunit4
+ testCompile libs.powermockEasymock
+ testCompile project(':clients').sourceSets.test.output
+
+ testRuntime libs.slf4jlog4j
+ }
+
+ javadoc {
+ enabled = false
+ }
+
+ tasks.create(name: "copyDependantLibs", type: Copy) {
+ from (configurations.testRuntime) {
+ include('slf4j-log4j12*')
+ include('log4j*jar')
+ }
+ from (configurations.runtime) {
+ exclude('kafka-clients*')
+ exclude('connect-*')
+ }
+ into "$buildDir/dependant-libs"
+ duplicatesStrategy 'exclude'
+ }
+
+ jar {
+ dependsOn copyDependantLibs
+ }
+}
+
task aggregatedJavadoc(type: Javadoc) {
def projectsWithJavadoc = subprojects.findAll { it.javadoc.enabled }
source = projectsWithJavadoc.collect { it.sourceSets.main.allJava }
diff --git a/checkstyle/checkstyle.xml b/checkstyle/checkstyle.xml
index ccab85ce4fecb..ad8545027a000 100644
--- a/checkstyle/checkstyle.xml
+++ b/checkstyle/checkstyle.xml
@@ -105,7 +105,7 @@
-
+
@@ -114,7 +114,7 @@
-
+
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 000acc38463b4..5549205bd1312 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -101,6 +101,9 @@
+
+
+
@@ -164,6 +167,8 @@
+
+
@@ -205,6 +210,8 @@
+
+
@@ -247,6 +254,10 @@
+
+
+
+
@@ -279,6 +290,7 @@
+
@@ -296,7 +308,16 @@
-
+
+
+
+
+
+
+
+
+
+
@@ -316,6 +337,8 @@
+
+
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index 45ee4e60a7c07..5bf69b6b65f9b 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -10,7 +10,7 @@
+ files="(Fetcher|Sender|SenderTest|ConsumerCoordinator|KafkaConsumer|KafkaProducer|Utils|TransactionManagerTest|KafkaAdminClient|NetworkClient|AdminClient).java"/>
+ files="(Errors|SaslAuthenticatorTest|AgentTest|CoordinatorTest).java"/>
@@ -57,7 +57,7 @@
files="AbstractRequest.java|KerberosLogin.java|WorkerSinkTaskTest.java|TransactionManagerTest.java"/>
+ files="(BufferPool|Fetcher|MetricName|Node|ConfigDef|RecordBatch|SslFactory|SslTransportLayer|MetadataResponse|KerberosLogin|Selector|Sender|Serdes|TokenInformation|Agent|Values|PluginUtils|MiniTrogdorCluster).java"/>
+
+
@@ -80,7 +83,7 @@
files="(KafkaConfigBackingStore|RequestResponseTest|WorkerSinkTaskTest).java"/>
+ files="(WorkerSinkTask|WorkerSourceTask).java"/>
-
-
-
-
-
+ files="(AbstractStatus|ConnectHeaders|ConnectRecord|ConnectSchema|DistributedHerder|FileStreamSourceTask|JsonConverter|KafkaConfigBackingStore).java"/>
+
@@ -155,11 +149,7 @@
files="StreamsPartitionAssignor.java"/>
-
-
+ files="(ProcessorStateManager|InternalTopologyBuilder|StreamsPartitionAssignor|StreamThread).java"/>
+ files="EosTestDriver|KStreamKStreamJoinTest.java|SmokeTestDriver.java"/>
@@ -212,6 +202,8 @@
files="SignalLogger.java"/>
+ now) {
+ return state.throttleUntilTimeMs - now;
+ } else {
+ return 0;
+ }
+ }
+
+ /**
+ * Return the number of milliseconds to wait, based on the connection state and the throttle time, before
+ * attempting to send data. If the connection has been established but being throttled, return throttle delay.
+ * Otherwise, return connection delay.
+ * @param id the connection to check
+ * @param now the current time in ms
+ */
+ public long pollDelayMs(String id, long now) {
+ long throttleDelayMs = throttleDelayMs(id, now);
+ if (isConnected(id) && throttleDelayMs > 0) {
+ return throttleDelayMs;
+ } else {
+ return connectionDelay(id, now);
+ }
+ }
+
/**
* Enter the checking_api_versions state for the given node.
* @param id the connection identifier
@@ -144,6 +187,7 @@ public void checkingApiVersions(String id) {
public void ready(String id) {
NodeConnectionState nodeState = nodeState(id);
nodeState.state = ConnectionState.READY;
+ nodeState.authenticationException = null;
resetReconnectBackoff(nodeState);
}
@@ -162,26 +206,43 @@ public void authenticationFailed(String id, long now, AuthenticationException ex
}
/**
- * Return true if the connection is ready.
+ * Return true if the connection is in the READY state and currently not throttled.
+ *
* @param id the connection identifier
+ * @param now the current time
*/
- public boolean isReady(String id) {
- NodeConnectionState state = nodeState.get(id);
- return state != null && state.state == ConnectionState.READY;
+ public boolean isReady(String id, long now) {
+ return isReady(nodeState.get(id), now);
+ }
+
+ private boolean isReady(NodeConnectionState state, long now) {
+ return state != null && state.state == ConnectionState.READY && state.throttleUntilTimeMs <= now;
}
/**
- * Return true if there is at least one node with connection in ready state and false otherwise.
+ * Return true if there is at least one node with connection in the READY state and not throttled. Returns false
+ * otherwise.
+ *
+ * @param now the current time
*/
- public boolean hasReadyNodes() {
+ public boolean hasReadyNodes(long now) {
for (Map.Entry entry : nodeState.entrySet()) {
- NodeConnectionState state = entry.getValue();
- if (state != null && state.state == ConnectionState.READY)
+ if (isReady(entry.getValue(), now)) {
return true;
+ }
}
return false;
}
+ /**
+ * Return true if the connection has been established
+ * @param id The id of the node to check
+ */
+ public boolean isConnected(String id) {
+ NodeConnectionState state = nodeState.get(id);
+ return state != null && state.state.isConnected();
+ }
+
/**
* Return true if the connection has been disconnected
* @param id The id of the node to check
@@ -271,6 +332,8 @@ private static class NodeConnectionState {
long lastConnectAttemptMs;
long failedAttempts;
long reconnectBackoffMs;
+ // Connection is being throttled if current time < throttleUntilTimeMs.
+ long throttleUntilTimeMs;
public NodeConnectionState(ConnectionState state, long lastConnectAttempt, long reconnectBackoffMs) {
this.state = state;
@@ -278,10 +341,11 @@ public NodeConnectionState(ConnectionState state, long lastConnectAttempt, long
this.lastConnectAttemptMs = lastConnectAttempt;
this.failedAttempts = 0;
this.reconnectBackoffMs = reconnectBackoffMs;
+ this.throttleUntilTimeMs = 0;
}
public String toString() {
- return "NodeState(" + state + ", " + lastConnectAttemptMs + ", " + failedAttempts + ")";
+ return "NodeState(" + state + ", " + lastConnectAttemptMs + ", " + failedAttempts + ", " + throttleUntilTimeMs + ")";
}
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/ConnectionState.java b/clients/src/main/java/org/apache/kafka/clients/ConnectionState.java
index 28b43d67fb1d7..f92c7fa037b0e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ConnectionState.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ConnectionState.java
@@ -31,4 +31,8 @@ public enum ConnectionState {
public boolean isDisconnected() {
return this == AUTHENTICATION_FAILED || this == DISCONNECTED;
}
+
+ public boolean isConnected() {
+ return this == CHECKING_API_VERSIONS || this == READY;
+ }
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java b/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
index a062818e3b2c4..5caee2d4c87d8 100644
--- a/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
+++ b/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
@@ -20,12 +20,12 @@
import java.util.Collections;
import java.util.Deque;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
-
/**
* The set of requests which have been sent or are being sent but haven't yet received a response
*/
@@ -151,9 +151,14 @@ public Iterable clearAll(String node) {
if (reqs == null) {
return Collections.emptyList();
} else {
- Deque clearedRequests = requests.remove(node);
+ final Deque clearedRequests = requests.remove(node);
inFlightRequestCount.getAndAdd(-clearedRequests.size());
- return clearedRequests;
+ return new Iterable() {
+ @Override
+ public Iterator iterator() {
+ return clearedRequests.descendingIterator();
+ }
+ };
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java b/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
index 0a9b51981e7ff..49bf3a3eab97b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
@@ -48,7 +48,7 @@ public interface KafkaClient extends Closeable {
boolean ready(Node node, long now);
/**
- * Returns the number of milliseconds to wait, based on the connection state, before attempting to send data. When
+ * Return the number of milliseconds to wait, based on the connection state, before attempting to send data. When
* disconnected, this respects the reconnect backoff time. When connecting or connected, this handles slow/stalled
* connections.
*
@@ -58,6 +58,16 @@ public interface KafkaClient extends Closeable {
*/
long connectionDelay(Node node, long now);
+ /**
+ * Return the number of milliseconds to wait, based on the connection state and the throttle time, before
+ * attempting to send data. If the connection has been established but being throttled, return throttle delay.
+ * Otherwise, return connection delay.
+ *
+ * @param node the connection to check
+ * @param now the current time in ms
+ */
+ long pollDelayMs(Node node, long now);
+
/**
* Check if the connection of the node has failed, based on the connection state. Such connection failure are
* usually transient and can be resumed in the next {@link #ready(org.apache.kafka.common.Node, long)} }
@@ -145,9 +155,12 @@ public interface KafkaClient extends Closeable {
boolean hasInFlightRequests(String nodeId);
/**
- * Return true if there is at least one node with connection in ready state and false otherwise.
+ * Return true if there is at least one node with connection in the READY state and not throttled. Returns false
+ * otherwise.
+ *
+ * @param now the current time
*/
- boolean hasReadyNodes();
+ boolean hasReadyNodes(long now);
/**
* Wake up the client if it is currently blocked waiting for I/O
diff --git a/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java b/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java
index 126728342d4c7..09ed995d14c3e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java
+++ b/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java
@@ -29,7 +29,7 @@
*
* This class is not thread-safe!
*/
-interface MetadataUpdater {
+public interface MetadataUpdater {
/**
* Gets the current cluster info without blocking.
diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
index 4b291b5eb67a4..26fa467f81a50 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -300,6 +300,22 @@ public long connectionDelay(Node node, long now) {
return connectionStates.connectionDelay(node.idString(), now);
}
+ // Return the remaining throttling delay in milliseconds if throttling is in progress. Return 0, otherwise.
+ // This is for testing.
+ public long throttleDelayMs(Node node, long now) {
+ return connectionStates.throttleDelayMs(node.idString(), now);
+ }
+
+ /**
+ * Return the poll delay in milliseconds based on both connection and throttle delay.
+ * @param node the connection to check
+ * @param now the current time in ms
+ */
+ @Override
+ public long pollDelayMs(Node node, long now) {
+ return connectionStates.pollDelayMs(node.idString(), now);
+ }
+
/**
* Check if the connection of the node has failed, based on the connection state. Such connection failure are
* usually transient and can be resumed in the next {@link #ready(org.apache.kafka.common.Node, long)} }
@@ -336,16 +352,18 @@ public AuthenticationException authenticationException(Node node) {
public boolean isReady(Node node, long now) {
// if we need to update our metadata now declare all requests unready to make metadata requests first
// priority
- return !metadataUpdater.isUpdateDue(now) && canSendRequest(node.idString());
+ return !metadataUpdater.isUpdateDue(now) && canSendRequest(node.idString(), now);
}
/**
* Are we connected and ready and able to send more requests to the given connection?
*
* @param node The node
+ * @param now the current timestamp
*/
- private boolean canSendRequest(String node) {
- return connectionStates.isReady(node) && selector.isChannelReady(node) && inFlightRequests.canSendMore(node);
+ private boolean canSendRequest(String node, long now) {
+ return connectionStates.isReady(node, now) && selector.isChannelReady(node) &&
+ inFlightRequests.canSendMore(node);
}
/**
@@ -373,7 +391,7 @@ private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long
// will be slightly different for some internal requests (for
// example, ApiVersionsRequests can be sent prior to being in
// READY state.)
- if (!canSendRequest(nodeId))
+ if (!canSendRequest(nodeId, now))
throw new IllegalStateException("Attempt to send a request to node " + nodeId + " which is not ready.");
}
AbstractRequest.Builder builder = clientRequest.requestBuilder();
@@ -513,8 +531,8 @@ public boolean hasInFlightRequests(String node) {
}
@Override
- public boolean hasReadyNodes() {
- return connectionStates.hasReadyNodes();
+ public boolean hasReadyNodes(long now) {
+ return connectionStates.hasReadyNodes(now);
}
/**
@@ -603,11 +621,13 @@ private void processDisconnection(List responses, String nodeId,
nodesNeedingApiVersionsFetch.remove(nodeId);
switch (disconnectState.state()) {
case AUTHENTICATION_FAILED:
- connectionStates.authenticationFailed(nodeId, now, disconnectState.exception());
- log.error("Connection to node {} failed authentication due to: {}", nodeId, disconnectState.exception().getMessage());
+ AuthenticationException exception = disconnectState.exception();
+ connectionStates.authenticationFailed(nodeId, now, exception);
+ metadataUpdater.handleAuthenticationFailure(exception);
+ log.error("Connection to node {} failed authentication due to: {}", nodeId, exception.getMessage());
break;
case AUTHENTICATE:
- // This warning applies to older brokers which dont provide feedback on authentication failures
+ // This warning applies to older brokers which don't provide feedback on authentication failures
log.warn("Connection to node {} terminated during authentication. This may indicate " +
"that authentication failed due to invalid credentials.", nodeId);
break;
@@ -625,9 +645,6 @@ private void processDisconnection(List responses, String nodeId,
else if (request.header.apiKey() == ApiKeys.METADATA)
metadataUpdater.handleDisconnection(request.destination);
}
- AuthenticationException authenticationException = connectionStates.authenticationException(nodeId);
- if (authenticationException != null)
- metadataUpdater.handleAuthenticationFailure(authenticationException);
}
/**
@@ -673,6 +690,24 @@ private void handleCompletedSends(List responses, long now) {
}
}
+ /**
+ * If a response from a node includes a non-zero throttle delay and client-side throttling has been enabled for
+ * the connection to the node, throttle the connection for the specified delay.
+ *
+ * @param response the response
+ * @param apiVersion the API version of the response
+ * @param nodeId the id of the node
+ * @param now The current time
+ */
+ private void maybeThrottle(AbstractResponse response, short apiVersion, String nodeId, long now) {
+ int throttleTimeMs = response.throttleTimeMs();
+ if (throttleTimeMs > 0 && response.shouldClientThrottle(apiVersion)) {
+ connectionStates.throttle(nodeId, now + throttleTimeMs);
+ log.trace("Connection to node {} is throttled for {} ms until timestamp {}", nodeId, throttleTimeMs,
+ now + throttleTimeMs);
+ }
+ }
+
/**
* Handle any completed receives and update the response list with the responses received.
*
@@ -689,7 +724,9 @@ private void handleCompletedReceives(List responses, long now) {
log.trace("Completed receive from node {} for {} with correlation id {}, received {}", req.destination,
req.header.apiKey(), req.header.correlationId(), responseStruct);
}
+ // If the received response includes a throttle delay, throttle the connection.
AbstractResponse body = AbstractResponse.parseResponse(req.header.apiKey(), responseStruct);
+ maybeThrottle(body, req.header.apiVersion(), req.destination, now);
if (req.isInternalRequest && body instanceof MetadataResponse)
metadataUpdater.handleCompletedMetadataResponse(req.header, now, (MetadataResponse) body);
else if (req.isInternalRequest && body instanceof ApiVersionsResponse)
@@ -716,9 +753,7 @@ private void handleApiVersionsResponse(List responses,
NodeApiVersions nodeVersionInfo = new NodeApiVersions(apiVersionsResponse.apiVersions());
apiVersions.update(node, nodeVersionInfo);
this.connectionStates.ready(node);
- if (log.isDebugEnabled()) {
- log.debug("Recorded API versions for node {}: {}", node, nodeVersionInfo);
- }
+ log.debug("Recorded API versions for node {}: {}", node, nodeVersionInfo);
}
/**
@@ -833,6 +868,7 @@ public long maybeUpdate(long now) {
long waitForMetadataFetch = this.metadataFetchInProgress ? requestTimeoutMs : 0;
long metadataTimeout = Math.max(timeToNextMetadataUpdate, waitForMetadataFetch);
+
if (metadataTimeout > 0) {
return metadataTimeout;
}
@@ -914,7 +950,7 @@ private boolean isAnyNodeConnecting() {
private long maybeUpdate(long now, Node node) {
String nodeConnectionId = node.idString();
- if (canSendRequest(nodeConnectionId)) {
+ if (canSendRequest(nodeConnectionId, now)) {
this.metadataFetchInProgress = true;
MetadataRequest.Builder metadataRequest;
if (metadata.needMetadataForAllTopics())
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StaleMetadataException.java b/clients/src/main/java/org/apache/kafka/clients/StaleMetadataException.java
similarity index 74%
rename from clients/src/main/java/org/apache/kafka/clients/consumer/internals/StaleMetadataException.java
rename to clients/src/main/java/org/apache/kafka/clients/StaleMetadataException.java
index 53110d3838747..dafc2d5483c4c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StaleMetadataException.java
+++ b/clients/src/main/java/org/apache/kafka/clients/StaleMetadataException.java
@@ -14,13 +14,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.kafka.clients.consumer.internals;
+package org.apache.kafka.clients;
import org.apache.kafka.common.errors.InvalidMetadataException;
/**
- * Thrown when metadata is old and needs to be refreshed.
+ * Thrown when current metadata cannot be used. This is often used as a way to trigger a metadata
+ * update before retrying another operation.
+ *
+ * Note: this is not a public API.
*/
public class StaleMetadataException extends InvalidMetadataException {
private static final long serialVersionUID = 1L;
+
+ public StaleMetadataException() {}
+
+ public StaleMetadataException(String message) {
+ super(message);
+ }
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
index 897e127d5577b..0171b617e7ca2 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
@@ -535,4 +535,237 @@ public DeleteRecordsResult deleteRecords(Map re
*/
public abstract DeleteRecordsResult deleteRecords(Map recordsToDelete,
DeleteRecordsOptions options);
+
+ /**
+ *
Create a Delegation Token.
+ *
+ *
This is a convenience method for {@link #createDelegationToken(CreateDelegationTokenOptions)} with default options.
+ * See the overload for more details.
This operation is supported by brokers with version 1.1.0 or higher.
+ *
+ *
The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the
+ * {@link CreateDelegationTokenResult#delegationToken() delegationToken()} method of the returned {@code CreateDelegationTokenResult}
+ *
+ *
{@link org.apache.kafka.common.errors.UnsupportedByAuthenticationException}
+ * If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
+ *
{@link org.apache.kafka.common.errors.InvalidPrincipalTypeException}
+ * if the renewers principal type is not supported.
+ *
{@link org.apache.kafka.common.errors.DelegationTokenDisabledException}
+ * if the delegation token feature is disabled.
+ *
{@link org.apache.kafka.common.errors.TimeoutException}
+ * if the request was not completed in within the given {@link CreateDelegationTokenOptions#timeoutMs()}.
+ *
+ *
+ * @param options The options to use when creating delegation token.
+ * @return The DeleteRecordsResult.
+ */
+ public abstract CreateDelegationTokenResult createDelegationToken(CreateDelegationTokenOptions options);
+
+
+ /**
+ *
Renew a Delegation Token.
+ *
+ *
This is a convenience method for {@link #renewDelegationToken(byte[], RenewDelegationTokenOptions)} with default options.
+ * See the overload for more details.
+ *
+ *
+ * @param hmac HMAC of the Delegation token
+ * @return The RenewDelegationTokenResult.
+ */
+ public RenewDelegationTokenResult renewDelegationToken(byte[] hmac) {
+ return renewDelegationToken(hmac, new RenewDelegationTokenOptions());
+ }
+
+ /**
+ *
Renew a Delegation Token.
+ *
+ *
This operation is supported by brokers with version 1.1.0 or higher.
+ *
+ *
The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the
+ * {@link RenewDelegationTokenResult#expiryTimestamp() expiryTimestamp()} method of the returned {@code RenewDelegationTokenResult}
+ *
+ *
{@link org.apache.kafka.common.errors.UnsupportedByAuthenticationException}
+ * If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
+ *
{@link org.apache.kafka.common.errors.DelegationTokenDisabledException}
+ * if the delegation token feature is disabled.
+ *
{@link org.apache.kafka.common.errors.DelegationTokenNotFoundException}
+ * if the delegation token is not found on server.
+ *
{@link org.apache.kafka.common.errors.DelegationTokenOwnerMismatchException}
+ * if the authenticated user is not owner/renewer of the token.
+ *
{@link org.apache.kafka.common.errors.DelegationTokenExpiredException}
+ * if the delegation token is expired.
+ *
{@link org.apache.kafka.common.errors.TimeoutException}
+ * if the request was not completed in within the given {@link RenewDelegationTokenOptions#timeoutMs()}.
+ *
+ *
+ * @param hmac HMAC of the Delegation token
+ * @param options The options to use when renewing delegation token.
+ * @return The RenewDelegationTokenResult.
+ */
+ public abstract RenewDelegationTokenResult renewDelegationToken(byte[] hmac, RenewDelegationTokenOptions options);
+
+ /**
+ *
Expire a Delegation Token.
+ *
+ *
This is a convenience method for {@link #expireDelegationToken(byte[], ExpireDelegationTokenOptions)} with default options.
+ * This will expire the token immediately. See the overload for more details.
+ *
+ * @param hmac HMAC of the Delegation token
+ * @return The ExpireDelegationTokenResult.
+ */
+ public ExpireDelegationTokenResult expireDelegationToken(byte[] hmac) {
+ return expireDelegationToken(hmac, new ExpireDelegationTokenOptions());
+ }
+
+ /**
+ *
Expire a Delegation Token.
+ *
+ *
This operation is supported by brokers with version 1.1.0 or higher.
+ *
+ *
The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the
+ * {@link ExpireDelegationTokenResult#expiryTimestamp() expiryTimestamp()} method of the returned {@code ExpireDelegationTokenResult}
+ *
+ *
{@link org.apache.kafka.common.errors.UnsupportedByAuthenticationException}
+ * If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
+ *
{@link org.apache.kafka.common.errors.DelegationTokenDisabledException}
+ * if the delegation token feature is disabled.
+ *
{@link org.apache.kafka.common.errors.DelegationTokenNotFoundException}
+ * if the delegation token is not found on server.
+ *
{@link org.apache.kafka.common.errors.DelegationTokenOwnerMismatchException}
+ * if the authenticated user is not owner/renewer of the requested token.
+ *
{@link org.apache.kafka.common.errors.DelegationTokenExpiredException}
+ * if the delegation token is expired.
+ *
{@link org.apache.kafka.common.errors.TimeoutException}
+ * if the request was not completed in within the given {@link ExpireDelegationTokenOptions#timeoutMs()}.
+ *
+ *
+ * @param hmac HMAC of the Delegation token
+ * @param options The options to use when expiring delegation token.
+ * @return The ExpireDelegationTokenResult.
+ */
+ public abstract ExpireDelegationTokenResult expireDelegationToken(byte[] hmac, ExpireDelegationTokenOptions options);
+
+ /**
+ *
Describe the Delegation Tokens.
+ *
+ *
This is a convenience method for {@link #describeDelegationToken(DescribeDelegationTokenOptions)} with default options.
+ * This will return all the user owned tokens and tokens where user have Describe permission. See the overload for more details.
This operation is supported by brokers with version 1.1.0 or higher.
+ *
+ *
The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the
+ * {@link DescribeDelegationTokenResult#delegationTokens() delegationTokens()} method of the returned {@code DescribeDelegationTokenResult}
+ *
+ *
{@link org.apache.kafka.common.errors.UnsupportedByAuthenticationException}
+ * If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
+ *
{@link org.apache.kafka.common.errors.DelegationTokenDisabledException}
+ * if the delegation token feature is disabled.
+ *
{@link org.apache.kafka.common.errors.TimeoutException}
+ * if the request was not completed in within the given {@link DescribeDelegationTokenOptions#timeoutMs()}.
+ *
+ *
+ * @param options The options to use when describing delegation tokens.
+ * @return The DescribeDelegationTokenResult.
+ */
+ public abstract DescribeDelegationTokenResult describeDelegationToken(DescribeDelegationTokenOptions options);
+
+ /**
+ * Describe some group IDs in the cluster.
+ *
+ * @param groupIds The IDs of the groups to describe.
+ * @param options The options to use when describing the groups.
+ * @return The DescribeConsumerGroupResult.
+ */
+ public abstract DescribeConsumerGroupsResult describeConsumerGroups(Collection groupIds,
+ DescribeConsumerGroupsOptions options);
+
+ /**
+ * Describe some group IDs in the cluster, with the default options.
+ *
+ * This is a convenience method for
+ * #{@link AdminClient#describeConsumerGroups(Collection, DescribeConsumerGroupsOptions)} with
+ * default options. See the overload for more details.
+ *
+ * @param groupIds The IDs of the groups to describe.
+ * @return The DescribeConsumerGroupResult.
+ */
+ public DescribeConsumerGroupsResult describeConsumerGroups(Collection groupIds) {
+ return describeConsumerGroups(groupIds, new DescribeConsumerGroupsOptions());
+ }
+
+ /**
+ * List the consumer groups available in the cluster.
+ *
+ * @param options The options to use when listing the consumer groups.
+ * @return The ListGroupsResult.
+ */
+ public abstract ListConsumerGroupsResult listConsumerGroups(ListConsumerGroupsOptions options);
+
+ /**
+ * List the consumer groups available in the cluster with the default options.
+ *
+ * This is a convenience method for #{@link AdminClient#listConsumerGroups(ListConsumerGroupsOptions)} with default options.
+ * See the overload for more details.
+ *
+ * @return The ListGroupsResult.
+ */
+ public ListConsumerGroupsResult listConsumerGroups() {
+ return listConsumerGroups(new ListConsumerGroupsOptions());
+ }
+
+ /**
+ * List the consumer group offsets available in the cluster.
+ *
+ * @param options The options to use when listing the consumer group offsets.
+ * @return The ListGroupOffsetsResult
+ */
+ public abstract ListConsumerGroupOffsetsResult listConsumerGroupOffsets(String groupId, ListConsumerGroupOffsetsOptions options);
+
+ /**
+ * List the consumer group offsets available in the cluster with the default options.
+ *
+ * This is a convenience method for #{@link AdminClient#listConsumerGroupOffsets(String, ListConsumerGroupOffsetsOptions)} with default options.
+ *
+ * @return The ListGroupOffsetsResult.
+ */
+ public ListConsumerGroupOffsetsResult listConsumerGroupOffsets(String groupId) {
+ return listConsumerGroupOffsets(groupId, new ListConsumerGroupOffsetsOptions());
+ }
+
+ /**
+ * Delete consumer groups from the cluster.
+ *
+ * @param options The options to use when deleting a consumer group.
+ * @return The DeletConsumerGroupResult.
+ */
+ public abstract DeleteConsumerGroupsResult deleteConsumerGroups(Collection groupIds, DeleteConsumerGroupsOptions options);
+
+ /**
+ * Delete consumer groups from the cluster with the default options.
+ *
+ * @return The DeleteConsumerGroupResult.
+ */
+ public DeleteConsumerGroupsResult deleteConsumerGroups(Collection groupIds) {
+ return deleteConsumerGroups(groupIds, new DeleteConsumerGroupsOptions());
+ }
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/Config.java b/clients/src/main/java/org/apache/kafka/clients/admin/Config.java
index 56f9c27079295..c81c0b60274b0 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/Config.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/Config.java
@@ -21,39 +21,40 @@
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
/**
* A configuration object containing the configuration entries for a resource.
- *
+ *
* The API of this class is evolving, see {@link AdminClient} for details.
*/
@InterfaceStability.Evolving
public class Config {
- private final Collection entries;
+ private final Map entries = new HashMap<>();
/**
* Create a configuration instance with the provided entries.
*/
public Config(Collection entries) {
- this.entries = Collections.unmodifiableCollection(entries);
+ for (ConfigEntry entry : entries) {
+ this.entries.put(entry.name(), entry);
+ }
}
/**
* Configuration entries for a resource.
*/
public Collection entries() {
- return entries;
+ return Collections.unmodifiableCollection(entries.values());
}
/**
* Get the configuration entry with the provided name or null if there isn't one.
*/
public ConfigEntry get(String name) {
- for (ConfigEntry entry : entries)
- if (entry.name().equals(name))
- return entry;
- return null;
+ return entries.get(name);
}
@Override
@@ -75,6 +76,6 @@ public int hashCode() {
@Override
public String toString() {
- return "Config(entries=" + entries + ")";
+ return "Config(entries=" + entries.values() + ")";
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupDescription.java b/clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupDescription.java
new file mode 100644
index 0000000000000..933d66ca4b761
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupDescription.java
@@ -0,0 +1,124 @@
+/*
+ * 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.kafka.clients.admin;
+
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.utils.Utils;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Objects;
+
+/**
+ * A detailed description of a single consumer group in the cluster.
+ */
+public class ConsumerGroupDescription {
+ private final String groupId;
+ private final boolean isSimpleConsumerGroup;
+ private final Collection members;
+ private final String partitionAssignor;
+ private final ConsumerGroupState state;
+ private final Node coordinator;
+
+ public ConsumerGroupDescription(String groupId,
+ boolean isSimpleConsumerGroup,
+ Collection members,
+ String partitionAssignor,
+ ConsumerGroupState state,
+ Node coordinator) {
+ this.groupId = groupId == null ? "" : groupId;
+ this.isSimpleConsumerGroup = isSimpleConsumerGroup;
+ this.members = members == null ? Collections.emptyList() :
+ Collections.unmodifiableList(new ArrayList<>(members));
+ this.partitionAssignor = partitionAssignor == null ? "" : partitionAssignor;
+ this.state = state;
+ this.coordinator = coordinator;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ ConsumerGroupDescription that = (ConsumerGroupDescription) o;
+ return isSimpleConsumerGroup == that.isSimpleConsumerGroup &&
+ groupId.equals(that.groupId) &&
+ members.equals(that.members) &&
+ partitionAssignor.equals(that.partitionAssignor) &&
+ state.equals(that.state);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(isSimpleConsumerGroup, groupId, members, partitionAssignor, state);
+ }
+
+ /**
+ * The id of the consumer group.
+ */
+ public String groupId() {
+ return groupId;
+ }
+
+ /**
+ * If consumer group is simple or not.
+ */
+ public boolean isSimpleConsumerGroup() {
+ return isSimpleConsumerGroup;
+ }
+
+ /**
+ * A list of the members of the consumer group.
+ */
+ public Collection members() {
+ return members;
+ }
+
+ /**
+ * The consumer group partition assignor.
+ */
+ public String partitionAssignor() {
+ return partitionAssignor;
+ }
+
+ /**
+ * The consumer group state, or UNKNOWN if the state is too new for us to parse.
+ */
+ public ConsumerGroupState state() {
+ return state;
+ }
+
+ /**
+ * The consumer group coordinator, or null if the coordinator is not known.
+ */
+ public Node coordinator() {
+ return coordinator;
+ }
+
+ @Override
+ public String toString() {
+ return "(groupId=" + groupId +
+ ", isSimpleConsumerGroup=" + isSimpleConsumerGroup +
+ ", members=" + Utils.join(members, ",") +
+ ", partitionAssignor=" + partitionAssignor +
+ ", state=" + state +
+ ", coordinator=" + coordinator +
+ ")";
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupListing.java b/clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupListing.java
new file mode 100644
index 0000000000000..46da9628010b6
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupListing.java
@@ -0,0 +1,59 @@
+/*
+ * 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.kafka.clients.admin;
+
+/**
+ * A listing of a consumer group in the cluster.
+ */
+public class ConsumerGroupListing {
+ private final String groupId;
+ private final boolean isSimpleConsumerGroup;
+
+ /**
+ * Create an instance with the specified parameters.
+ *
+ * @param groupId Group Id
+ * @param isSimpleConsumerGroup If consumer group is simple or not.
+ */
+ public ConsumerGroupListing(String groupId, boolean isSimpleConsumerGroup) {
+ this.groupId = groupId;
+ this.isSimpleConsumerGroup = isSimpleConsumerGroup;
+ }
+
+ /**
+ * Consumer Group Id
+ */
+ public String groupId() {
+ return groupId;
+ }
+
+ /**
+ * If Consumer Group is simple or not.
+ */
+ public boolean isSimpleConsumerGroup() {
+ return isSimpleConsumerGroup;
+ }
+
+ @Override
+ public String toString() {
+ return "(" +
+ "groupId='" + groupId + '\'' +
+ ", isSimpleConsumerGroup=" + isSimpleConsumerGroup +
+ ')';
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenOptions.java
new file mode 100644
index 0000000000000..1b77b943800ec
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenOptions.java
@@ -0,0 +1,53 @@
+/*
+ * 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.kafka.clients.admin;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+
+/**
+ * Options for {@link AdminClient#createDelegationToken(CreateDelegationTokenOptions)}.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class CreateDelegationTokenOptions extends AbstractOptions {
+ private long maxLifeTimeMs = -1;
+ private List renewers = new LinkedList<>();
+
+ public CreateDelegationTokenOptions renewers(List renewers) {
+ this.renewers = renewers;
+ return this;
+ }
+
+ public List renewers() {
+ return renewers;
+ }
+
+ public CreateDelegationTokenOptions maxlifeTimeMs(long maxLifeTimeMs) {
+ this.maxLifeTimeMs = maxLifeTimeMs;
+ return this;
+ }
+
+ public long maxlifeTimeMs() {
+ return maxLifeTimeMs;
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenResult.java
new file mode 100644
index 0000000000000..043cbe87fef5c
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenResult.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.security.token.delegation.DelegationToken;
+
+/**
+ * The result of the {@link KafkaAdminClient#createDelegationToken(CreateDelegationTokenOptions)} call.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class CreateDelegationTokenResult {
+ private final KafkaFuture delegationToken;
+
+ CreateDelegationTokenResult(KafkaFuture delegationToken) {
+ this.delegationToken = delegationToken;
+ }
+
+ /**
+ * Returns a future which yields a delegation token
+ */
+ public KafkaFuture delegationToken() {
+ return delegationToken;
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteConsumerGroupsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteConsumerGroupsOptions.java
new file mode 100644
index 0000000000000..cd505f4c11138
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteConsumerGroupsOptions.java
@@ -0,0 +1,31 @@
+/*
+ * 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.kafka.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collection;
+
+/**
+ * Options for the {@link AdminClient#deleteConsumerGroups(Collection)} call.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class DeleteConsumerGroupsOptions extends AbstractOptions {
+
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteConsumerGroupsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteConsumerGroupsResult.java
new file mode 100644
index 0000000000000..dd6835cf10cbc
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteConsumerGroupsResult.java
@@ -0,0 +1,52 @@
+/*
+ * 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.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * The result of the {@link AdminClient#deleteConsumerGroups(Collection)} call.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class DeleteConsumerGroupsResult {
+ private final Map> futures;
+
+ DeleteConsumerGroupsResult(final Map> futures) {
+ this.futures = futures;
+ }
+
+ /**
+ * Return a map from group id to futures which can be used to check the status of
+ * individual deletions.
+ */
+ public Map> deletedGroups() {
+ return futures;
+ }
+
+ /**
+ * Return a future which succeeds only if all the consumer group deletions succeed.
+ */
+ public KafkaFuture all() {
+ return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0]));
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeConsumerGroupsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeConsumerGroupsOptions.java
new file mode 100644
index 0000000000000..7daff1a483be8
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeConsumerGroupsOptions.java
@@ -0,0 +1,31 @@
+/*
+ * 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.kafka.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collection;
+
+/**
+ * Options for {@link AdminClient#describeConsumerGroups(Collection, DescribeConsumerGroupsOptions)}.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class DescribeConsumerGroupsOptions extends AbstractOptions {
+}
\ No newline at end of file
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeConsumerGroupsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeConsumerGroupsResult.java
new file mode 100644
index 0000000000000..8f0ebad4f7ac6
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeConsumerGroupsResult.java
@@ -0,0 +1,72 @@
+/*
+ * 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.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+
+/**
+ * The result of the {@link KafkaAdminClient#describeConsumerGroups(Collection, DescribeConsumerGroupsOptions)}} call.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class DescribeConsumerGroupsResult {
+
+ private final Map> futures;
+
+ public DescribeConsumerGroupsResult(final Map> futures) {
+ this.futures = futures;
+ }
+
+ /**
+ * Return a map from group id to futures which yield group descriptions.
+ */
+ public Map> describedGroups() {
+ return futures;
+ }
+
+ /**
+ * Return a future which yields all ConsumerGroupDescription objects, if all the describes succeed.
+ */
+ public KafkaFuture> all() {
+ return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])).thenApply(
+ new KafkaFuture.BaseFunction>() {
+ @Override
+ public Map apply(Void v) {
+ try {
+ Map descriptions = new HashMap<>(futures.size());
+ for (Map.Entry> entry : futures.entrySet()) {
+ descriptions.put(entry.getKey(), entry.getValue().get());
+ }
+ return descriptions;
+ } catch (InterruptedException | ExecutionException e) {
+ // This should be unreachable, since the KafkaFuture#allOf already ensured
+ // that all of the futures completed successfully.
+ throw new RuntimeException(e);
+ }
+ }
+ });
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenOptions.java
new file mode 100644
index 0000000000000..60b99354e3552
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenOptions.java
@@ -0,0 +1,48 @@
+/*
+ * 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.kafka.clients.admin;
+
+import java.util.List;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+
+/**
+ * Options for {@link AdminClient#describeDelegationToken(DescribeDelegationTokenOptions)}.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class DescribeDelegationTokenOptions extends AbstractOptions {
+ private List owners;
+
+ /**
+ * if owners is null, all the user owned tokens and tokens where user have Describe permission
+ * will be returned.
+ * @param owners
+ * @return this instance
+ */
+ public DescribeDelegationTokenOptions owners(List owners) {
+ this.owners = owners;
+ return this;
+ }
+
+ public List owners() {
+ return owners;
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenResult.java
new file mode 100644
index 0000000000000..7a9d4b9dd9737
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenResult.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import java.util.List;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.security.token.delegation.DelegationToken;
+
+/**
+ * The result of the {@link KafkaAdminClient#describeDelegationToken(DescribeDelegationTokenOptions)} call.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class DescribeDelegationTokenResult {
+ private final KafkaFuture> delegationTokens;
+
+ DescribeDelegationTokenResult(KafkaFuture> delegationTokens) {
+ this.delegationTokens = delegationTokens;
+ }
+
+ /**
+ * Returns a future which yields list of delegation tokens
+ */
+ public KafkaFuture> delegationTokens() {
+ return delegationTokens;
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java
new file mode 100644
index 0000000000000..138cd4e69e477
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+/**
+ * Options for {@link AdminClient#expireDelegationToken(byte[], ExpireDelegationTokenOptions)}.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class ExpireDelegationTokenOptions extends AbstractOptions {
+ private long expiryTimePeriodMs = -1L;
+
+ public ExpireDelegationTokenOptions expiryTimePeriodMs(long expiryTimePeriodMs) {
+ this.expiryTimePeriodMs = expiryTimePeriodMs;
+ return this;
+ }
+
+ public long expiryTimePeriodMs() {
+ return expiryTimePeriodMs;
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenResult.java
new file mode 100644
index 0000000000000..41782bdcb5ce7
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenResult.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+/**
+ * The result of the {@link KafkaAdminClient#expireDelegationToken(byte[], ExpireDelegationTokenOptions)} call.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class ExpireDelegationTokenResult {
+ private final KafkaFuture expiryTimestamp;
+
+ ExpireDelegationTokenResult(KafkaFuture expiryTimestamp) {
+ this.expiryTimestamp = expiryTimestamp;
+ }
+
+ /**
+ * Returns a future which yields expiry timestamp
+ */
+ public KafkaFuture expiryTimestamp() {
+ return expiryTimestamp;
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 36cbe6cad0032..6b02e11e9bbc2 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -22,12 +22,17 @@
import org.apache.kafka.clients.ClientResponse;
import org.apache.kafka.clients.ClientUtils;
import org.apache.kafka.clients.KafkaClient;
-import org.apache.kafka.clients.Metadata;
import org.apache.kafka.clients.NetworkClient;
+import org.apache.kafka.clients.StaleMetadataException;
import org.apache.kafka.clients.admin.DeleteAclsResult.FilterResult;
import org.apache.kafka.clients.admin.DeleteAclsResult.FilterResults;
import org.apache.kafka.clients.admin.DescribeReplicaLogDirsResult.ReplicaLogDirInfo;
+import org.apache.kafka.clients.admin.internals.AdminMetadataManager;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.clients.consumer.internals.PartitionAssignor;
import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.ConsumerGroupState;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;
@@ -41,8 +46,8 @@
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.errors.AuthenticationException;
-import org.apache.kafka.common.errors.BrokerNotAvailableException;
import org.apache.kafka.common.errors.DisconnectException;
+import org.apache.kafka.common.errors.InvalidGroupIdException;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.InvalidTopicException;
import org.apache.kafka.common.errors.RetriableException;
@@ -69,6 +74,8 @@
import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation;
import org.apache.kafka.common.requests.CreateAclsResponse;
import org.apache.kafka.common.requests.CreateAclsResponse.AclCreationResponse;
+import org.apache.kafka.common.requests.CreateDelegationTokenRequest;
+import org.apache.kafka.common.requests.CreateDelegationTokenResponse;
import org.apache.kafka.common.requests.CreatePartitionsRequest;
import org.apache.kafka.common.requests.CreatePartitionsResponse;
import org.apache.kafka.common.requests.CreateTopicsRequest;
@@ -77,6 +84,8 @@
import org.apache.kafka.common.requests.DeleteAclsResponse;
import org.apache.kafka.common.requests.DeleteAclsResponse.AclDeletionResult;
import org.apache.kafka.common.requests.DeleteAclsResponse.AclFilterResponse;
+import org.apache.kafka.common.requests.DeleteGroupsRequest;
+import org.apache.kafka.common.requests.DeleteGroupsResponse;
import org.apache.kafka.common.requests.DeleteRecordsRequest;
import org.apache.kafka.common.requests.DeleteRecordsResponse;
import org.apache.kafka.common.requests.DeleteTopicsRequest;
@@ -85,12 +94,28 @@
import org.apache.kafka.common.requests.DescribeAclsResponse;
import org.apache.kafka.common.requests.DescribeConfigsRequest;
import org.apache.kafka.common.requests.DescribeConfigsResponse;
+import org.apache.kafka.common.requests.DescribeDelegationTokenRequest;
+import org.apache.kafka.common.requests.DescribeDelegationTokenResponse;
+import org.apache.kafka.common.requests.DescribeGroupsRequest;
+import org.apache.kafka.common.requests.DescribeGroupsResponse;
import org.apache.kafka.common.requests.DescribeLogDirsRequest;
import org.apache.kafka.common.requests.DescribeLogDirsResponse;
+import org.apache.kafka.common.requests.ExpireDelegationTokenRequest;
+import org.apache.kafka.common.requests.ExpireDelegationTokenResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.FindCoordinatorResponse;
+import org.apache.kafka.common.requests.ListGroupsRequest;
+import org.apache.kafka.common.requests.ListGroupsResponse;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.requests.RenewDelegationTokenRequest;
+import org.apache.kafka.common.requests.RenewDelegationTokenResponse;
import org.apache.kafka.common.requests.Resource;
import org.apache.kafka.common.requests.ResourceType;
+import org.apache.kafka.common.security.token.delegation.DelegationToken;
+import org.apache.kafka.common.security.token.delegation.TokenInformation;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.common.utils.KafkaThread;
import org.apache.kafka.common.utils.LogContext;
@@ -158,9 +183,9 @@ public class KafkaAdminClient extends AdminClient {
private final Time time;
/**
- * The cluster metadata used by the KafkaClient.
+ * The cluster metadata manager used by the KafkaClient.
*/
- private final Metadata metadata;
+ private final AdminMetadataManager metadataManager;
/**
* The metrics for this KafkaAdminClient.
@@ -195,6 +220,8 @@ public class KafkaAdminClient extends AdminClient {
private final int maxRetries;
+ private final long retryBackoffMs;
+
/**
* Get or create a list value from a map.
*
@@ -299,8 +326,9 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso
try {
// Since we only request node information, it's safe to pass true for allowAutoTopicCreation (and it
// simplifies communication with older brokers)
- Metadata metadata = new Metadata(config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG),
- config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG), true);
+ AdminMetadataManager metadataManager = new AdminMetadataManager(logContext,
+ config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG),
+ config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG));
List reporters = config.getConfiguredInstances(AdminClientConfig.METRIC_REPORTER_CLASSES_CONFIG,
MetricsReporter.class);
Map metricTags = Collections.singletonMap("client-id", clientId);
@@ -316,7 +344,7 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso
metrics, time, metricGrpPrefix, channelBuilder, logContext);
networkClient = new NetworkClient(
selector,
- metadata,
+ metadataManager.updater(),
clientId,
1,
config.getLong(AdminClientConfig.RECONNECT_BACKOFF_MS_CONFIG),
@@ -328,7 +356,7 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso
true,
apiVersions,
logContext);
- return new KafkaAdminClient(config, clientId, time, metadata, metrics, networkClient,
+ return new KafkaAdminClient(config, clientId, time, metadataManager, metrics, networkClient,
timeoutProcessorFactory, logContext);
} catch (Throwable exc) {
closeQuietly(metrics, "Metrics");
@@ -339,35 +367,44 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso
}
}
- static KafkaAdminClient createInternal(AdminClientConfig config, KafkaClient client, Metadata metadata, Time time) {
+ static KafkaAdminClient createInternal(AdminClientConfig config, KafkaClient client, Time time) {
Metrics metrics = null;
String clientId = generateClientId(config);
try {
metrics = new Metrics(new MetricConfig(), new LinkedList(), time);
- return new KafkaAdminClient(config, clientId, time, metadata, metrics, client, null,
- createLogContext(clientId));
+ LogContext logContext = createLogContext(clientId);
+ AdminMetadataManager metadataManager = new AdminMetadataManager(logContext,
+ config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG),
+ config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG));
+ return new KafkaAdminClient(config, clientId, time, metadataManager, metrics,
+ client, null, logContext);
} catch (Throwable exc) {
closeQuietly(metrics, "Metrics");
throw new KafkaException("Failed create new KafkaAdminClient", exc);
}
}
- private static LogContext createLogContext(String clientId) {
+ static LogContext createLogContext(String clientId) {
return new LogContext("[AdminClient clientId=" + clientId + "] ");
}
- private KafkaAdminClient(AdminClientConfig config, String clientId, Time time, Metadata metadata,
- Metrics metrics, KafkaClient client, TimeoutProcessorFactory timeoutProcessorFactory,
- LogContext logContext) {
+ private KafkaAdminClient(AdminClientConfig config,
+ String clientId,
+ Time time,
+ AdminMetadataManager metadataManager,
+ Metrics metrics,
+ KafkaClient client,
+ TimeoutProcessorFactory timeoutProcessorFactory,
+ LogContext logContext) {
this.defaultTimeoutMs = config.getInt(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG);
this.clientId = clientId;
this.log = logContext.logger(KafkaAdminClient.class);
this.time = time;
- this.metadata = metadata;
+ this.metadataManager = metadataManager;
List addresses = ClientUtils.parseAndValidateAddresses(
config.getList(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG));
- this.metadata.update(Cluster.bootstrap(addresses), Collections.emptySet(), time.milliseconds());
+ metadataManager.update(Cluster.bootstrap(addresses), time.milliseconds());
this.metrics = metrics;
this.client = client;
this.runnable = new AdminClientRunnable();
@@ -376,6 +413,7 @@ private KafkaAdminClient(AdminClientConfig config, String clientId, Time time, M
this.timeoutProcessorFactory = (timeoutProcessorFactory == null) ?
new TimeoutProcessorFactory() : timeoutProcessorFactory;
this.maxRetries = config.getInt(AdminClientConfig.RETRIES_CONFIG);
+ this.retryBackoffMs = config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG);
config.logUnused();
AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics);
log.debug("Kafka admin client initialized");
@@ -434,6 +472,13 @@ private interface NodeProvider {
Node provide();
}
+ private class MetadataUpdateNodeIdProvider implements NodeProvider {
+ @Override
+ public Node provide() {
+ return client.leastLoadedNode(time.milliseconds());
+ }
+ }
+
private class ConstantNodeIdProvider implements NodeProvider {
private final int nodeId;
@@ -443,7 +488,16 @@ private class ConstantNodeIdProvider implements NodeProvider {
@Override
public Node provide() {
- return metadata.fetch().nodeById(nodeId);
+ if (metadataManager.isReady() &&
+ (metadataManager.nodeById(nodeId) != null)) {
+ return metadataManager.nodeById(nodeId);
+ }
+ // If we can't find the node with the given constant ID, we schedule a
+ // metadata update and hope it appears. This behavior is useful for avoiding
+ // flaky behavior in tests when the cluster is starting up and not all nodes
+ // have appeared.
+ metadataManager.requestUpdate();
+ return null;
}
}
@@ -453,7 +507,12 @@ public Node provide() {
private class ControllerNodeProvider implements NodeProvider {
@Override
public Node provide() {
- return metadata.fetch().controller();
+ if (metadataManager.isReady() &&
+ (metadataManager.controller() != null)) {
+ return metadataManager.controller();
+ }
+ metadataManager.requestUpdate();
+ return null;
}
}
@@ -463,23 +522,41 @@ public Node provide() {
private class LeastLoadedNodeProvider implements NodeProvider {
@Override
public Node provide() {
- return client.leastLoadedNode(time.milliseconds());
+ if (metadataManager.isReady()) {
+ // This may return null if all nodes are busy.
+ // In that case, we will postpone node assignment.
+ return client.leastLoadedNode(time.milliseconds());
+ }
+ metadataManager.requestUpdate();
+ return null;
}
}
abstract class Call {
+ private final boolean internal;
private final String callName;
private final long deadlineMs;
private final NodeProvider nodeProvider;
private int tries = 0;
private boolean aborted = false;
+ private Node curNode = null;
+ private long nextAllowedTryMs = 0;
- Call(String callName, long deadlineMs, NodeProvider nodeProvider) {
+ Call(boolean internal, String callName, long deadlineMs, NodeProvider nodeProvider) {
+ this.internal = internal;
this.callName = callName;
this.deadlineMs = deadlineMs;
this.nodeProvider = nodeProvider;
}
+ Call(String callName, long deadlineMs, NodeProvider nodeProvider) {
+ this(false, callName, deadlineMs, nodeProvider);
+ }
+
+ protected Node curNode() {
+ return curNode;
+ }
+
/**
* Handle a failure.
*
@@ -508,11 +585,13 @@ final void fail(long now, Throwable throwable) {
// this RPC. That is why 'tries' is not incremented.
if ((throwable instanceof UnsupportedVersionException) &&
handleUnsupportedVersionException((UnsupportedVersionException) throwable)) {
- log.trace("{} attempting protocol downgrade.", this);
+ log.debug("{} attempting protocol downgrade and then retry.", this);
runnable.enqueue(this, now);
return;
}
tries++;
+ nextAllowedTryMs = now + retryBackoffMs;
+
// If the call has timed out, fail.
if (calcTimeoutMsRemainingAsInt(now, deadlineMs) < 0) {
if (log.isDebugEnabled()) {
@@ -587,6 +666,10 @@ boolean handleUnsupportedVersionException(UnsupportedVersionException exception)
public String toString() {
return "Call(callName=" + callName + ", deadlineMs=" + deadlineMs + ")";
}
+
+ public boolean isInternal() {
+ return internal;
+ }
}
static class TimeoutProcessorFactory {
@@ -664,58 +747,52 @@ int nextTimeoutMs() {
private final class AdminClientRunnable implements Runnable {
/**
- * Pending calls. Protected by the object monitor.
- * This will be null only if the thread has shut down.
+ * Calls which have not yet been assigned to a node.
+ * Only accessed from this thread.
*/
- private List newCalls = new LinkedList<>();
+ private final ArrayList pendingCalls = new ArrayList<>();
/**
- * Check if the AdminClient metadata is ready.
- * We need to know who the controller is, and have a non-empty view of the cluster.
- *
- * @param prevMetadataVersion The previous metadata version which wasn't usable.
- * @return null if the metadata is usable; the current metadata
- * version otherwise
+ * Maps nodes to calls that we want to send.
+ * Only accessed from this thread.
*/
- private Integer checkMetadataReady(Integer prevMetadataVersion) {
- if (prevMetadataVersion != null) {
- if (prevMetadataVersion == metadata.version())
- return prevMetadataVersion;
- }
- Cluster cluster = metadata.fetch();
- if (cluster.nodes().isEmpty()) {
- log.trace("Metadata is not ready yet. No cluster nodes found.");
- return metadata.requestUpdate();
- }
- if (cluster.controller() == null) {
- log.trace("Metadata is not ready yet. No controller found.");
- return metadata.requestUpdate();
- }
- if (prevMetadataVersion != null) {
- log.trace("Metadata is now ready.");
- }
- return null;
- }
+ private final Map> callsToSend = new HashMap<>();
+
+ /**
+ * Maps node ID strings to calls that have been sent.
+ * Only accessed from this thread.
+ */
+ private final Map> callsInFlight = new HashMap<>();
+
+ /**
+ * Maps correlation IDs to calls that have been sent.
+ * Only accessed from this thread.
+ */
+ private final Map correlationIdToCalls = new HashMap<>();
+
+ /**
+ * Pending calls. Protected by the object monitor.
+ * This will be null only if the thread has shut down.
+ */
+ private List newCalls = new LinkedList<>();
/**
- * Time out the elements in the newCalls list which are expired.
+ * Time out the elements in the pendingCalls list which are expired.
*
* @param processor The timeout processor.
*/
- private synchronized void timeoutNewCalls(TimeoutProcessor processor) {
- int numTimedOut = processor.handleTimeouts(newCalls,
- "Timed out waiting for a node assignment.");
+ private void timeoutPendingCalls(TimeoutProcessor processor) {
+ int numTimedOut = processor.handleTimeouts(pendingCalls, "Timed out waiting for a node assignment.");
if (numTimedOut > 0)
- log.debug("Timed out {} new calls.", numTimedOut);
+ log.debug("Timed out {} pending calls.", numTimedOut);
}
/**
* Time out calls which have been assigned to nodes.
*
* @param processor The timeout processor.
- * @param callsToSend A map of nodes to the calls they need to handle.
*/
- private void timeoutCallsToSend(TimeoutProcessor processor, Map> callsToSend) {
+ private int timeoutCallsToSend(TimeoutProcessor processor) {
int numTimedOut = 0;
for (List callList : callsToSend.values()) {
numTimedOut += processor.handleTimeouts(callList,
@@ -723,65 +800,70 @@ private void timeoutCallsToSend(TimeoutProcessor processor, Map
}
if (numTimedOut > 0)
log.debug("Timed out {} call(s) with assigned nodes.", numTimedOut);
+ return numTimedOut;
}
/**
- * Choose nodes for the calls in the callsToSend list.
+ * Drain all the calls from newCalls into pendingCalls.
*
* This function holds the lock for the minimum amount of time, to avoid blocking
* users of AdminClient who will also take the lock to add new calls.
- *
- * @param now The current time in milliseconds.
- * @param callsToSend A map of nodes to the calls they need to handle.
- *
*/
- private void chooseNodesForNewCalls(long now, Map> callsToSend) {
- List newCallsToAdd = null;
- synchronized (this) {
- if (newCalls.isEmpty()) {
- return;
- }
- newCallsToAdd = newCalls;
- newCalls = new LinkedList<>();
- }
- for (Call call : newCallsToAdd) {
- chooseNodeForNewCall(now, callsToSend, call);
+ private synchronized void drainNewCalls() {
+ if (!newCalls.isEmpty()) {
+ pendingCalls.addAll(newCalls);
+ newCalls.clear();
}
}
/**
- * Choose a node for a new call.
+ * Choose nodes for the calls in the pendingCalls list.
*
* @param now The current time in milliseconds.
- * @param callsToSend A map of nodes to the calls they need to handle.
- * @param call The call.
+ * @param pendingIter An iterator yielding pending calls.
*/
- private void chooseNodeForNewCall(long now, Map> callsToSend, Call call) {
- Node node = call.nodeProvider.provide();
- if (node == null) {
- call.fail(now, new BrokerNotAvailableException(
- String.format("Error choosing node for %s: no node found.", call.callName)));
- return;
+ private long chooseNodesForPendingCalls(long now, Iterator pendingIter) {
+ long pollTimeout = Long.MAX_VALUE;
+ log.trace("Trying to choose nodes for {} at {}", pendingIter, now);
+ while (pendingIter.hasNext()) {
+ Call call = pendingIter.next();
+
+ // If the call is being retried, await the proper backoff before finding the node
+ if (now < call.nextAllowedTryMs) {
+ pollTimeout = Math.min(pollTimeout, call.nextAllowedTryMs - now);
+ continue;
+ }
+
+ Node node = null;
+ try {
+ node = call.nodeProvider.provide();
+ } catch (Throwable t) {
+ // Handle authentication errors while choosing nodes.
+ log.debug("Unable to choose node for {}", call, t);
+ pendingIter.remove();
+ call.fail(now, t);
+ }
+ if (node != null) {
+ log.trace("Assigned {} to node {}", call, node);
+ pendingIter.remove();
+ call.curNode = node;
+ getOrCreateListValue(callsToSend, node).add(call);
+ } else {
+ log.trace("Unable to assign {} to a node.", call);
+ }
}
- log.trace("Assigned {} to {}", call, node);
- getOrCreateListValue(callsToSend, node).add(call);
+ return pollTimeout;
}
/**
* Send the calls which are ready.
*
* @param now The current time in milliseconds.
- * @param callsToSend The calls to send, by node.
- * @param correlationIdToCalls A map of correlation IDs to calls.
- * @param callsInFlight A map of nodes to the calls they have in flight.
- *
* @return The minimum timeout we need for poll().
*/
- private long sendEligibleCalls(long now, Map> callsToSend,
- Map correlationIdToCalls, Map> callsInFlight) {
+ private long sendEligibleCalls(long now) {
long pollTimeout = Long.MAX_VALUE;
- for (Iterator>> iter = callsToSend.entrySet().iterator();
- iter.hasNext(); ) {
+ for (Iterator>> iter = callsToSend.entrySet().iterator(); iter.hasNext(); ) {
Map.Entry> entry = iter.next();
List calls = entry.getValue();
if (calls.isEmpty()) {
@@ -790,7 +872,7 @@ private long sendEligibleCalls(long now, Map> callsToSend,
}
Node node = entry.getKey();
if (!client.ready(node, now)) {
- long nodeTimeout = client.connectionDelay(node, now);
+ long nodeTimeout = client.pollDelayMs(node, now);
pollTimeout = Math.min(pollTimeout, nodeTimeout);
log.trace("Client is not ready to send to {}. Must delay {} ms", node, nodeTimeout);
continue;
@@ -822,9 +904,8 @@ private long sendEligibleCalls(long now, Map> callsToSend,
* to time them out is to close the entire connection.
*
* @param processor The timeout processor.
- * @param callsInFlight A map of nodes to the calls they have in flight.
*/
- private void timeoutCallsInFlight(TimeoutProcessor processor, Map> callsInFlight) {
+ private void timeoutCallsInFlight(TimeoutProcessor processor) {
int numTimedOut = 0;
for (Map.Entry> entry : callsInFlight.entrySet()) {
List contexts = entry.getValue();
@@ -852,51 +933,17 @@ private void timeoutCallsInFlight(TimeoutProcessor processor, Map> callsToSend) {
- AuthenticationException authenticationException = metadata.getAndClearAuthenticationException();
- if (authenticationException == null) {
- for (Node node : callsToSend.keySet()) {
- authenticationException = client.authenticationException(node);
- if (authenticationException != null)
- break;
- }
- }
- if (authenticationException != null) {
- synchronized (this) {
- failCalls(now, newCalls, authenticationException);
- }
- for (List calls : callsToSend.values()) {
- failCalls(now, calls, authenticationException);
- }
- callsToSend.clear();
- }
- }
-
- private void failCalls(long now, List calls, AuthenticationException authenticationException) {
- for (Call call : calls) {
- call.fail(now, authenticationException);
- }
- calls.clear();
- }
-
/**
* Handle responses from the server.
*
* @param now The current time in milliseconds.
* @param responses The latest responses from KafkaClient.
- * @param correlationIdToCall A map of correlation IDs to calls.
- * @param callsInFlight A map of nodes to the calls they have in flight.
- **/
- private void handleResponses(long now, List responses, Map> callsInFlight,
- Map correlationIdToCall) {
+ **/
+ private void handleResponses(long now, List responses) {
for (ClientResponse response : responses) {
int correlationId = response.requestHeader().correlationId();
- Call call = correlationIdToCall.get(correlationId);
+ Call call = correlationIdToCalls.get(correlationId);
if (call == null) {
// If the server returns information about a correlation ID we didn't use yet,
// an internal server error has occurred. Close the connection and log an error message.
@@ -908,7 +955,7 @@ private void handleResponses(long now, List responses, Map calls = callsInFlight.get(response.destination());
if ((calls == null) || (!calls.remove(call))) {
log.error("Internal server error on {}: ignoring call {} in correlationIdToCall " +
@@ -921,9 +968,14 @@ private void handleResponses(long now, List responses, Map responses, Map> callsToSend, Map correlationIdToCalls) {
- if (newCalls.isEmpty() && callsToSend.isEmpty() && correlationIdToCalls.isEmpty()) {
+ /**
+ * Reassign calls that have not yet been sent. When metadata is refreshed,
+ * all unsent calls are reassigned to handle controller change and node changes.
+ * When a node is disconnected, all calls assigned to the node are reassigned.
+ *
+ * @param now The current time in milliseconds
+ * @param disconnectedOnly Reassign only calls to nodes that were disconnected
+ * in the last poll
+ */
+ private void reassignUnsentCalls(long now, boolean disconnectedOnly) {
+ ArrayList pendingCallsToSend = new ArrayList<>();
+ for (Iterator>> iter = callsToSend.entrySet().iterator(); iter.hasNext(); ) {
+ Map.Entry> entry = iter.next();
+ if (!disconnectedOnly || client.connectionFailed(entry.getKey())) {
+ for (Call call : entry.getValue()) {
+ pendingCallsToSend.add(call);
+ }
+ iter.remove();
+ }
+ }
+ chooseNodesForPendingCalls(now, pendingCallsToSend.iterator());
+ }
+
+ private boolean hasActiveExternalCalls(Collection calls) {
+ for (Call call : calls) {
+ if (!call.isInternal()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Return true if there are currently active external calls.
+ */
+ private boolean hasActiveExternalCalls() {
+ if (hasActiveExternalCalls(pendingCalls)) {
+ return true;
+ }
+ for (List callList : callsToSend.values()) {
+ if (hasActiveExternalCalls(callList)) {
+ return true;
+ }
+ }
+ return hasActiveExternalCalls(correlationIdToCalls.values());
+ }
+
+ private boolean threadShouldExit(long now, long curHardShutdownTimeMs) {
+ if (!hasActiveExternalCalls()) {
log.trace("All work has been completed, and the I/O thread is now exiting.");
return true;
}
- if (now > curHardShutdownTimeMs) {
+ if (now >= curHardShutdownTimeMs) {
log.info("Forcing a hard I/O thread shutdown. Requests in progress will be aborted.");
return true;
}
@@ -955,52 +1053,42 @@ private synchronized boolean threadShouldExit(long now, long curHardShutdownTime
@Override
public void run() {
- /*
- * Maps nodes to calls that we want to send.
- */
- Map> callsToSend = new HashMap<>();
-
- /*
- * Maps node ID strings to calls that have been sent.
- */
- Map> callsInFlight = new HashMap<>();
-
- /*
- * Maps correlation IDs to calls that have been sent.
- */
- Map correlationIdToCalls = new HashMap<>();
-
- /*
- * The previous metadata version which wasn't usable, or null if there is none.
- */
- Integer prevMetadataVersion = null;
-
long now = time.milliseconds();
log.trace("Thread starting");
while (true) {
+ // Copy newCalls into pendingCalls.
+ drainNewCalls();
+
// Check if the AdminClient thread should shut down.
long curHardShutdownTimeMs = hardShutdownTimeMs.get();
- if ((curHardShutdownTimeMs != INVALID_SHUTDOWN_TIME) &&
- threadShouldExit(now, curHardShutdownTimeMs, callsToSend, correlationIdToCalls))
+ if ((curHardShutdownTimeMs != INVALID_SHUTDOWN_TIME) && threadShouldExit(now, curHardShutdownTimeMs))
break;
// Handle timeouts.
TimeoutProcessor timeoutProcessor = timeoutProcessorFactory.create(now);
- timeoutNewCalls(timeoutProcessor);
- timeoutCallsToSend(timeoutProcessor, callsToSend);
- timeoutCallsInFlight(timeoutProcessor, callsInFlight);
+ timeoutPendingCalls(timeoutProcessor);
+ timeoutCallsToSend(timeoutProcessor);
+ timeoutCallsInFlight(timeoutProcessor);
long pollTimeout = Math.min(1200000, timeoutProcessor.nextTimeoutMs());
if (curHardShutdownTimeMs != INVALID_SHUTDOWN_TIME) {
pollTimeout = Math.min(pollTimeout, curHardShutdownTimeMs - now);
}
- // Handle new calls and metadata update requests.
- prevMetadataVersion = checkMetadataReady(prevMetadataVersion);
- if (prevMetadataVersion == null) {
- chooseNodesForNewCalls(now, callsToSend);
- pollTimeout = Math.min(pollTimeout,
- sendEligibleCalls(now, callsToSend, correlationIdToCalls, callsInFlight));
+ // Choose nodes for our pending calls.
+ pollTimeout = Math.min(pollTimeout, chooseNodesForPendingCalls(now, pendingCalls.iterator()));
+ long metadataFetchDelayMs = metadataManager.metadataFetchDelayMs(now);
+ if (metadataFetchDelayMs == 0) {
+ metadataManager.transitionToUpdatePending(now);
+ Call metadataCall = makeMetadataCall(now);
+ // Create a new metadata fetch call and add it to the end of pendingCalls.
+ // Assign a node for just the new call (we handled the other pending nodes above).
+ pendingCalls.add(metadataCall);
+ chooseNodesForPendingCalls(now, pendingCalls.listIterator(pendingCalls.size() - 1));
+ }
+ pollTimeout = Math.min(pollTimeout, sendEligibleCalls(now));
+ if (metadataFetchDelayMs > 0) {
+ pollTimeout = Math.min(pollTimeout, metadataFetchDelayMs);
}
// Wait for network responses.
@@ -1010,20 +1098,21 @@ public void run() {
// Update the current time and handle the latest responses.
now = time.milliseconds();
- handleAuthenticationException(now, callsToSend);
- handleResponses(now, responses, callsInFlight, correlationIdToCalls);
+ reassignUnsentCalls(now, true); // reassign calls to disconnected nodes
+ handleResponses(now, responses);
}
int numTimedOut = 0;
TimeoutProcessor timeoutProcessor = new TimeoutProcessor(Long.MAX_VALUE);
synchronized (this) {
- numTimedOut += timeoutProcessor.handleTimeouts(newCalls,
- "The AdminClient thread has exited.");
+ numTimedOut += timeoutProcessor.handleTimeouts(newCalls, "The AdminClient thread has exited.");
newCalls = null;
}
+ numTimedOut += timeoutProcessor.handleTimeouts(pendingCalls, "The AdminClient thread has exited.");
+ numTimedOut += timeoutCallsToSend(timeoutProcessor);
numTimedOut += timeoutProcessor.handleTimeouts(correlationIdToCalls.values(),
"The AdminClient thread has exited.");
if (numTimedOut > 0) {
- log.debug("Timed out {} remaining operations.", numTimedOut);
+ log.debug("Timed out {} remaining operation(s).", numTimedOut);
}
closeQuietly(client, "KafkaClient");
closeQuietly(metrics, "Metrics");
@@ -1075,6 +1164,35 @@ void call(Call call, long now) {
enqueue(call, now);
}
}
+
+ /**
+ * Create a new metadata call.
+ */
+ private Call makeMetadataCall(long now) {
+ return new Call(true, "fetchMetadata", calcDeadlineMs(now, defaultTimeoutMs),
+ new MetadataUpdateNodeIdProvider()) {
+ @Override
+ public AbstractRequest.Builder createRequest(int timeoutMs) {
+ // Since this only requests node information, it's safe to pass true
+ // for allowAutoTopicCreation (and it simplifies communication with
+ // older brokers)
+ return new MetadataRequest.Builder(Collections.emptyList(), true);
+ }
+
+ @Override
+ public void handleResponse(AbstractResponse abstractResponse) {
+ MetadataResponse response = (MetadataResponse) abstractResponse;
+ long now = time.milliseconds();
+ metadataManager.update(response.cluster(), now);
+ reassignUnsentCalls(now, false);
+ }
+
+ @Override
+ public void handleFailure(Throwable e) {
+ metadataManager.updateFailed(e);
+ }
+ };
+ }
}
/**
@@ -1083,7 +1201,11 @@ void call(Call call, long now) {
* those policies on the server, so that they can be changed in the future if needed.
*/
private static boolean topicNameIsUnrepresentable(String topicName) {
- return (topicName == null) || topicName.isEmpty();
+ return topicName == null || topicName.isEmpty();
+ }
+
+ private static boolean groupIdIsUnrepresentable(String groupId) {
+ return groupId == null;
}
@Override
@@ -1114,6 +1236,14 @@ public AbstractRequest.Builder createRequest(int timeoutMs) {
@Override
public void handleResponse(AbstractResponse abstractResponse) {
CreateTopicsResponse response = (CreateTopicsResponse) abstractResponse;
+ // Check for controller change
+ for (ApiError error : response.errors().values()) {
+ if (error.error() == Errors.NOT_CONTROLLER) {
+ metadataManager.clearController();
+ metadataManager.requestUpdate();
+ throw error.exception();
+ }
+ }
// Handle server responses for particular topics.
for (Map.Entry entry : response.errors().entrySet()) {
KafkaFutureImpl future = topicFutures.get(entry.getKey());
@@ -1150,9 +1280,10 @@ void handleFailure(Throwable throwable) {
}
@Override
- public DeleteTopicsResult deleteTopics(final Collection topicNames,
+ public DeleteTopicsResult deleteTopics(Collection topicNames,
DeleteTopicsOptions options) {
final Map> topicFutures = new HashMap<>(topicNames.size());
+ final List validTopicNames = new ArrayList<>(topicNames.size());
for (String topicName : topicNames) {
if (topicNameIsUnrepresentable(topicName)) {
KafkaFutureImpl future = new KafkaFutureImpl<>();
@@ -1161,6 +1292,7 @@ public DeleteTopicsResult deleteTopics(final Collection topicNames,
topicFutures.put(topicName, future);
} else if (!topicFutures.containsKey(topicName)) {
topicFutures.put(topicName, new KafkaFutureImpl());
+ validTopicNames.add(topicName);
}
}
final long now = time.milliseconds();
@@ -1169,12 +1301,20 @@ public DeleteTopicsResult deleteTopics(final Collection topicNames,
@Override
AbstractRequest.Builder createRequest(int timeoutMs) {
- return new DeleteTopicsRequest.Builder(new HashSet<>(topicNames), timeoutMs);
+ return new DeleteTopicsRequest.Builder(new HashSet<>(validTopicNames), timeoutMs);
}
@Override
void handleResponse(AbstractResponse abstractResponse) {
DeleteTopicsResponse response = (DeleteTopicsResponse) abstractResponse;
+ // Check for controller change
+ for (Errors error : response.errors().values()) {
+ if (error == Errors.NOT_CONTROLLER) {
+ metadataManager.clearController();
+ metadataManager.requestUpdate();
+ throw error.exception();
+ }
+ }
// Handle server responses for particular topics.
for (Map.Entry entry : response.errors().entrySet()) {
KafkaFutureImpl future = topicFutures.get(entry.getKey());
@@ -1204,7 +1344,7 @@ void handleFailure(Throwable throwable) {
completeAllExceptionally(topicFutures.values(), throwable);
}
};
- if (!topicNames.isEmpty()) {
+ if (!validTopicNames.isEmpty()) {
runnable.call(call, now);
}
return new DeleteTopicsResult(new HashMap>(topicFutures));
@@ -1276,15 +1416,16 @@ AbstractRequest.Builder createRequest(int timeoutMs) {
void handleResponse(AbstractResponse abstractResponse) {
MetadataResponse response = (MetadataResponse) abstractResponse;
// Handle server responses for particular topics.
+ Cluster cluster = response.cluster();
+ Map errors = response.errors();
for (Map.Entry> entry : topicFutures.entrySet()) {
String topicName = entry.getKey();
KafkaFutureImpl future = entry.getValue();
- Errors topicError = response.errors().get(topicName);
+ Errors topicError = errors.get(topicName);
if (topicError != null) {
future.completeExceptionally(topicError.exception());
continue;
}
- Cluster cluster = response.cluster();
if (!cluster.topics().contains(topicName)) {
future.completeExceptionally(new InvalidTopicException("Topic " + topicName + " not found."));
continue;
@@ -1923,6 +2064,7 @@ void handleFailure(Throwable throwable) {
return new DescribeReplicaLogDirsResult(new HashMap>(futures));
}
+ @Override
public CreatePartitionsResult createPartitions(Map newPartitions,
final CreatePartitionsOptions options) {
final Map> futures = new HashMap<>(newPartitions.size());
@@ -1943,6 +2085,14 @@ public AbstractRequest.Builder createRequest(int timeoutMs) {
@Override
public void handleResponse(AbstractResponse abstractResponse) {
CreatePartitionsResponse response = (CreatePartitionsResponse) abstractResponse;
+ // Check for controller change
+ for (ApiError error : response.errors().values()) {
+ if (error.error() == Errors.NOT_CONTROLLER) {
+ metadataManager.clearController();
+ metadataManager.requestUpdate();
+ throw error.exception();
+ }
+ }
for (Map.Entry result : response.errors().entrySet()) {
KafkaFutureImpl future = futures.get(result.getKey());
if (result.getValue().isSuccess()) {
@@ -1961,6 +2111,7 @@ void handleFailure(Throwable throwable) {
return new CreatePartitionsResult(new HashMap>(futures));
}
+ @Override
public DeleteRecordsResult deleteRecords(final Map recordsToDelete,
final DeleteRecordsOptions options) {
@@ -2069,4 +2220,513 @@ void handleFailure(Throwable throwable) {
return new DeleteRecordsResult(new HashMap>(futures));
}
+
+ @Override
+ public CreateDelegationTokenResult createDelegationToken(final CreateDelegationTokenOptions options) {
+ final KafkaFutureImpl delegationTokenFuture = new KafkaFutureImpl<>();
+ final long now = time.milliseconds();
+ runnable.call(new Call("createDelegationToken", calcDeadlineMs(now, options.timeoutMs()),
+ new LeastLoadedNodeProvider()) {
+
+ @Override
+ AbstractRequest.Builder createRequest(int timeoutMs) {
+ return new CreateDelegationTokenRequest.Builder(options.renewers(), options.maxlifeTimeMs());
+ }
+
+ @Override
+ void handleResponse(AbstractResponse abstractResponse) {
+ CreateDelegationTokenResponse response = (CreateDelegationTokenResponse) abstractResponse;
+ if (response.hasError()) {
+ delegationTokenFuture.completeExceptionally(response.error().exception());
+ } else {
+ TokenInformation tokenInfo = new TokenInformation(response.tokenId(), response.owner(),
+ options.renewers(), response.issueTimestamp(), response.maxTimestamp(), response.expiryTimestamp());
+ DelegationToken token = new DelegationToken(tokenInfo, response.hmacBytes());
+ delegationTokenFuture.complete(token);
+ }
+ }
+
+ @Override
+ void handleFailure(Throwable throwable) {
+ delegationTokenFuture.completeExceptionally(throwable);
+ }
+ }, now);
+
+ return new CreateDelegationTokenResult(delegationTokenFuture);
+ }
+
+ @Override
+ public RenewDelegationTokenResult renewDelegationToken(final byte[] hmac, final RenewDelegationTokenOptions options) {
+ final KafkaFutureImpl expiryTimeFuture = new KafkaFutureImpl<>();
+ final long now = time.milliseconds();
+ runnable.call(new Call("renewDelegationToken", calcDeadlineMs(now, options.timeoutMs()),
+ new LeastLoadedNodeProvider()) {
+
+ @Override
+ AbstractRequest.Builder createRequest(int timeoutMs) {
+ return new RenewDelegationTokenRequest.Builder(hmac, options.renewTimePeriodMs());
+ }
+
+ @Override
+ void handleResponse(AbstractResponse abstractResponse) {
+ RenewDelegationTokenResponse response = (RenewDelegationTokenResponse) abstractResponse;
+ if (response.hasError()) {
+ expiryTimeFuture.completeExceptionally(response.error().exception());
+ } else {
+ expiryTimeFuture.complete(response.expiryTimestamp());
+ }
+ }
+
+ @Override
+ void handleFailure(Throwable throwable) {
+ expiryTimeFuture.completeExceptionally(throwable);
+ }
+ }, now);
+
+ return new RenewDelegationTokenResult(expiryTimeFuture);
+ }
+
+ @Override
+ public ExpireDelegationTokenResult expireDelegationToken(final byte[] hmac, final ExpireDelegationTokenOptions options) {
+ final KafkaFutureImpl expiryTimeFuture = new KafkaFutureImpl<>();
+ final long now = time.milliseconds();
+ runnable.call(new Call("expireDelegationToken", calcDeadlineMs(now, options.timeoutMs()),
+ new LeastLoadedNodeProvider()) {
+
+ @Override
+ AbstractRequest.Builder createRequest(int timeoutMs) {
+ return new ExpireDelegationTokenRequest.Builder(hmac, options.expiryTimePeriodMs());
+ }
+
+ @Override
+ void handleResponse(AbstractResponse abstractResponse) {
+ ExpireDelegationTokenResponse response = (ExpireDelegationTokenResponse) abstractResponse;
+ if (response.hasError()) {
+ expiryTimeFuture.completeExceptionally(response.error().exception());
+ } else {
+ expiryTimeFuture.complete(response.expiryTimestamp());
+ }
+ }
+
+ @Override
+ void handleFailure(Throwable throwable) {
+ expiryTimeFuture.completeExceptionally(throwable);
+ }
+ }, now);
+
+ return new ExpireDelegationTokenResult(expiryTimeFuture);
+ }
+
+ @Override
+ public DescribeDelegationTokenResult describeDelegationToken(final DescribeDelegationTokenOptions options) {
+ final KafkaFutureImpl> tokensFuture = new KafkaFutureImpl<>();
+ final long now = time.milliseconds();
+ runnable.call(new Call("describeDelegationToken", calcDeadlineMs(now, options.timeoutMs()),
+ new LeastLoadedNodeProvider()) {
+
+ @Override
+ AbstractRequest.Builder createRequest(int timeoutMs) {
+ return new DescribeDelegationTokenRequest.Builder(options.owners());
+ }
+
+ @Override
+ void handleResponse(AbstractResponse abstractResponse) {
+ DescribeDelegationTokenResponse response = (DescribeDelegationTokenResponse) abstractResponse;
+ if (response.hasError()) {
+ tokensFuture.completeExceptionally(response.error().exception());
+ } else {
+ tokensFuture.complete(response.tokens());
+ }
+ }
+
+ @Override
+ void handleFailure(Throwable throwable) {
+ tokensFuture.completeExceptionally(throwable);
+ }
+ }, now);
+
+ return new DescribeDelegationTokenResult(tokensFuture);
+ }
+
+ @Override
+ public DescribeConsumerGroupsResult describeConsumerGroups(final Collection groupIds,
+ final DescribeConsumerGroupsOptions options) {
+
+ final Map> futures = new HashMap<>(groupIds.size());
+ for (String groupId: groupIds) {
+ if (groupIdIsUnrepresentable(groupId)) {
+ KafkaFutureImpl future = new KafkaFutureImpl<>();
+ future.completeExceptionally(new InvalidGroupIdException("The given group id '" +
+ groupId + "' cannot be represented in a request."));
+ futures.put(groupId, future);
+ } else if (!futures.containsKey(groupId)) {
+ futures.put(groupId, new KafkaFutureImpl());
+ }
+ }
+
+ // TODO: KAFKA-6788, we should consider grouping the request per coordinator and send one request with a list of
+ // all consumer groups this coordinator host
+ for (final Map.Entry> entry : futures.entrySet()) {
+ // skip sending request for those futures that already failed.
+ if (entry.getValue().isCompletedExceptionally())
+ continue;
+
+ final String groupId = entry.getKey();
+
+ final long startFindCoordinatorMs = time.milliseconds();
+ final long deadline = calcDeadlineMs(startFindCoordinatorMs, options.timeoutMs());
+
+ runnable.call(new Call("findCoordinator", deadline, new LeastLoadedNodeProvider()) {
+ @Override
+ AbstractRequest.Builder createRequest(int timeoutMs) {
+ return new FindCoordinatorRequest.Builder(FindCoordinatorRequest.CoordinatorType.GROUP, groupId);
+ }
+
+ @Override
+ void handleResponse(AbstractResponse abstractResponse) {
+ final FindCoordinatorResponse fcResponse = (FindCoordinatorResponse) abstractResponse;
+ Errors error = fcResponse.error();
+ if (error == Errors.COORDINATOR_NOT_AVAILABLE) {
+ // Retry COORDINATOR_NOT_AVAILABLE, in case the error is temporary.
+ throw error.exception();
+ } else if (error != Errors.NONE) {
+ // All other errors are immediate failures.
+ KafkaFutureImpl future = futures.get(groupId);
+ future.completeExceptionally(error.exception());
+ return;
+ }
+
+ final long nowDescribeConsumerGroups = time.milliseconds();
+ final int nodeId = fcResponse.node().id();
+ runnable.call(new Call("describeConsumerGroups", deadline, new ConstantNodeIdProvider(nodeId)) {
+ @Override
+ AbstractRequest.Builder createRequest(int timeoutMs) {
+ return new DescribeGroupsRequest.Builder(Collections.singletonList(groupId));
+ }
+
+ @Override
+ void handleResponse(AbstractResponse abstractResponse) {
+ final DescribeGroupsResponse response = (DescribeGroupsResponse) abstractResponse;
+
+ KafkaFutureImpl future = futures.get(groupId);
+ final DescribeGroupsResponse.GroupMetadata groupMetadata = response.groups().get(groupId);
+
+ final Errors groupError = groupMetadata.error();
+ if (groupError != Errors.NONE) {
+ // TODO: KAFKA-6789, we can retry based on the error code
+ future.completeExceptionally(groupError.exception());
+ } else {
+ final String protocolType = groupMetadata.protocolType();
+ if (protocolType.equals(ConsumerProtocol.PROTOCOL_TYPE) || protocolType.isEmpty()) {
+ final List members = groupMetadata.members();
+ final List memberDescriptions = new ArrayList<>(members.size());
+
+ for (DescribeGroupsResponse.GroupMember groupMember : members) {
+ Set partitions = Collections.emptySet();
+ if (groupMember.memberAssignment().remaining() > 0) {
+ final PartitionAssignor.Assignment assignment = ConsumerProtocol.
+ deserializeAssignment(groupMember.memberAssignment().duplicate());
+ partitions = new HashSet<>(assignment.partitions());
+ }
+ final MemberDescription memberDescription =
+ new MemberDescription(groupMember.memberId(),
+ groupMember.clientId(),
+ groupMember.clientHost(),
+ new MemberAssignment(partitions));
+ memberDescriptions.add(memberDescription);
+ }
+ final ConsumerGroupDescription consumerGroupDescription =
+ new ConsumerGroupDescription(groupId,
+ protocolType.isEmpty(),
+ memberDescriptions,
+ groupMetadata.protocol(),
+ ConsumerGroupState.parse(groupMetadata.state()),
+ fcResponse.node());
+ future.complete(consumerGroupDescription);
+ }
+ }
+ }
+
+ @Override
+ void handleFailure(Throwable throwable) {
+ KafkaFutureImpl future = futures.get(groupId);
+ future.completeExceptionally(throwable);
+ }
+ }, nowDescribeConsumerGroups);
+ }
+
+ @Override
+ void handleFailure(Throwable throwable) {
+ KafkaFutureImpl future = futures.get(groupId);
+ future.completeExceptionally(throwable);
+ }
+ }, startFindCoordinatorMs);
+ }
+
+ return new DescribeConsumerGroupsResult(new HashMap>(futures));
+ }
+
+ private final static class ListConsumerGroupsResults {
+ private final List errors;
+ private final HashMap listings;
+ private final HashSet remaining;
+ private final KafkaFutureImpl> future;
+
+ ListConsumerGroupsResults(Collection leaders,
+ KafkaFutureImpl> future) {
+ this.errors = new ArrayList<>();
+ this.listings = new HashMap<>();
+ this.remaining = new HashSet<>(leaders);
+ this.future = future;
+ tryComplete();
+ }
+
+ synchronized void addError(Throwable throwable, Node node) {
+ ApiError error = ApiError.fromThrowable(throwable);
+ if (error.message() == null || error.message().isEmpty()) {
+ errors.add(error.error().exception("Error listing groups on " + node));
+ } else {
+ errors.add(error.error().exception("Error listing groups on " + node + ": " + error.message()));
+ }
+ }
+
+ synchronized void addListing(ConsumerGroupListing listing) {
+ listings.put(listing.groupId(), listing);
+ }
+
+ synchronized void tryComplete(Node leader) {
+ remaining.remove(leader);
+ tryComplete();
+ }
+
+ private synchronized void tryComplete() {
+ if (remaining.isEmpty()) {
+ ArrayList