From ea9f8eed4c98134f06720b9649492db9ed744241 Mon Sep 17 00:00:00 2001 From: Luke Chen <43372967+showuon@users.noreply.github.com> Date: Tue, 22 Sep 2020 04:35:16 +0800 Subject: [PATCH 01/15] KAFKA-10401; Ensure `currentStateTimeStamp` is set correctly by group coordinator (#9202) Fix the `currentStateTimeStamp` doesn't get set in `GROUP_METADATA_VALUE_SCHEMA_V3`, and did a small refactor to use the `GROUP_VALUE_SCHEMAS.size - 1` replace the default hard-coded max version number. Also add test for it. Reviewers: Jason Gustafson , Ismael Juma --- .../group/GroupMetadataManager.scala | 18 ++++---- .../group/GroupMetadataManagerTest.scala | 41 ++++++++++++++++++- 2 files changed, 47 insertions(+), 12 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala index 785dcc37a9bc1..76383a2224028 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala @@ -1118,6 +1118,7 @@ object GroupMetadataManager { private val CURRENT_OFFSET_KEY_SCHEMA = schemaForKey(CURRENT_OFFSET_KEY_SCHEMA_VERSION) private val CURRENT_GROUP_KEY_SCHEMA = schemaForKey(CURRENT_GROUP_KEY_SCHEMA_VERSION) + private val CURRENT_GROUP_METADATA_VALUE_SCHEMA_VERSION = GROUP_VALUE_SCHEMAS.keySet.max private def schemaForKey(version: Int) = { val schemaOpt = MESSAGE_TYPE_SCHEMAS.get(version) @@ -1381,23 +1382,18 @@ object GroupMetadataManager { val valueSchema = schemaForGroupValue(version) val value = valueSchema.read(buffer) - if (version >= 0 && version <= 3) { + if (version >= 0 && version <= CURRENT_GROUP_METADATA_VALUE_SCHEMA_VERSION) { val generationId = value.get(GENERATION_KEY).asInstanceOf[Int] val protocolType = value.get(PROTOCOL_TYPE_KEY).asInstanceOf[String] val protocol = value.get(PROTOCOL_KEY).asInstanceOf[String] val leaderId = value.get(LEADER_KEY).asInstanceOf[String] val memberMetadataArray = value.getArray(MEMBERS_KEY) val initialState = if (memberMetadataArray.isEmpty) Empty else Stable - val currentStateTimestamp: Option[Long] = version match { - case version if version == 2 => - if (value.hasField(CURRENT_STATE_TIMESTAMP_KEY)) { - val timestamp = value.getLong(CURRENT_STATE_TIMESTAMP_KEY) - if (timestamp == -1) None else Some(timestamp) - } else - None - case _ => - None - } + val currentStateTimestamp: Option[Long] = + if (version >= 2 && value.hasField(CURRENT_STATE_TIMESTAMP_KEY)) { + val timestamp = value.getLong(CURRENT_STATE_TIMESTAMP_KEY) + if (timestamp == -1) None else Some(timestamp) + } else None val members = memberMetadataArray.map { memberMetadataObj => val memberMetadata = memberMetadataObj.asInstanceOf[Struct] diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala index 3e3fb66e95a58..af34b9e5e1f1d 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala @@ -43,8 +43,9 @@ import org.apache.kafka.common.record._ import org.apache.kafka.common.requests.OffsetFetchResponse import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse import org.apache.kafka.common.utils.Utils +import org.apache.kafka.common.KafkaException import org.easymock.{Capture, EasyMock, IAnswer} -import org.junit.Assert.{assertEquals, assertFalse, assertNull, assertTrue} +import org.junit.Assert.{assertEquals, assertFalse, assertNull, assertTrue, assertThrows} import org.junit.{Before, Test} import org.scalatest.Assertions.fail @@ -908,6 +909,44 @@ class GroupMetadataManagerTest { assertTrue(group.has(memberId)) } + @Test + def testShouldThrowExceptionForUnsupportedGroupMetadataVersion(): Unit = { + val generation = 1 + val protocol = "range" + val memberId = "memberId" + val unsupportedVersion = Short.MinValue + + // put the unsupported version as the version value + val groupMetadataRecordValue = buildStableGroupRecordWithMember(generation, protocolType, protocol, memberId) + .value().putShort(unsupportedVersion) + // reset the position to the starting position 0 so that it can read the data in correct order + groupMetadataRecordValue.position(0) + + val e = assertThrows(classOf[KafkaException], + () => GroupMetadataManager.readGroupMessageValue(groupId, groupMetadataRecordValue, time)) + assertEquals(s"Unknown group metadata version ${unsupportedVersion}", e.getMessage) + } + + @Test + def testCurrentStateTimestampForAllGroupMetadataVersions(): Unit = { + val generation = 1 + val protocol = "range" + val memberId = "memberId" + + for (apiVersion <- ApiVersion.allVersions) { + val groupMetadataRecord = buildStableGroupRecordWithMember(generation, protocolType, protocol, memberId, apiVersion = apiVersion) + + val deserializedGroupMetadata = GroupMetadataManager.readGroupMessageValue(groupId, groupMetadataRecord.value(), time) + // GROUP_METADATA_VALUE_SCHEMA_V2 or higher should correctly set the currentStateTimestamp + if (apiVersion >= KAFKA_2_1_IV0) + assertEquals(s"the apiVersion $apiVersion doesn't set the currentStateTimestamp correctly.", + Some(time.milliseconds()), deserializedGroupMetadata.currentStateTimestamp) + else + assertTrue(s"the apiVersion $apiVersion should not set the currentStateTimestamp.", + deserializedGroupMetadata.currentStateTimestamp.isEmpty) + } + } + @Test def testReadFromOldGroupMetadata(): Unit = { val generation = 1 From ef0f2d96f93c502cfc33062426721bf8e2807d45 Mon Sep 17 00:00:00 2001 From: Micah Paul Ramos Date: Mon, 28 Sep 2020 12:56:36 -0700 Subject: [PATCH 02/15] KAFKA-9584: Fix Headers ConcurrentModificationException in Streams (#8181) Avoid forwarding a shared reference to the record context in punctuate calls. Note, this fix isn't airtight, since all processors triggered by a single punctuate call will still see the same reference to the record context. It's also not a terribly principled approach, since the context is still technically not defined, but this is about the best we can do without significant refactoring. We will probably follow up with a more comprehensive solution, but this should avoid the issue for most programs. Reviewers: Matthias J. Sax , John Roesler --- .../streams/processor/internals/StreamTask.java | 4 ++-- .../processor/internals/StreamTaskTest.java | 14 ++++++++++++++ 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java index 9aa8e793571e8..c809cbe16855a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java @@ -68,7 +68,6 @@ */ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator { - private static final ConsumerRecord DUMMY_RECORD = new ConsumerRecord<>(ProcessorContextImpl.NONEXIST_TOPIC, -1, -1L, null, null); // visible for testing static final byte LATEST_MAGIC_BYTE = 1; @@ -435,7 +434,8 @@ public void punctuate(final ProcessorNode node, final long timestamp, final Punc throw new IllegalStateException(String.format("%sCurrent node is not null", logPrefix)); } - updateProcessorContext(new StampedRecord(DUMMY_RECORD, timestamp), node); + updateProcessorContext(new StampedRecord(new ConsumerRecord<>(ProcessorContextImpl.NONEXIST_TOPIC, -1, -1L, null, null), + timestamp), node); if (log.isTraceEnabled()) { log.trace("Punctuating processor {} with timestamp {} and punctuation type {}", node.name(), timestamp, type); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java index 2832291951f94..0d38231e21d8c 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java @@ -1131,6 +1131,20 @@ public void shouldFlushRecordCollectorOnFlushState() { assertTrue(collector.flushed()); } + @Test + public void shouldNotShareHeadersBetweenPunctuateIterations() { + task = createStatelessTask(createConfig(false), StreamsConfig.METRICS_LATEST); + task.initializeMetadata(); + task.initializeTopology(); + + task.punctuate(processorSystemTime, 1, PunctuationType.WALL_CLOCK_TIME, timestamp -> { + task.processorContext.recordContext().headers().add("dummy", (byte[]) null); + }); + task.punctuate(processorSystemTime, 1, PunctuationType.WALL_CLOCK_TIME, timestamp -> { + assertFalse(task.processorContext.recordContext().headers().iterator().hasNext()); + }); + } + @Test public void shouldCheckpointOffsetsOnCommit() throws IOException { task = createStatefulTask(createConfig(false), true); From a8e49ef394b549b91b75a6625a60e02a44718c49 Mon Sep 17 00:00:00 2001 From: John Roesler Date: Fri, 24 Jul 2020 15:30:20 -0500 Subject: [PATCH 03/15] MINOR: add task ':streams:testAll' (#9073) Adds a new task (`:streams:testAll`) to gradle to run all the tests for all Streams sub-projects. Reviewers: Boyang Chen --- README.md | 4 ++++ build.gradle | 35 +++++++++++++++++++++++++++++------ 2 files changed, 33 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 07153fb202669..d09b469c19995 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,10 @@ This is for `core`, `examples` and `clients` ./gradlew core:jar ./gradlew core:test +Streams has multiple sub-projects, but you can run all the tests: + + ./gradlew :streams:testAll + ### Listing all gradle tasks ### ./gradlew tasks diff --git a/build.gradle b/build.gradle index 6879aa55d438d..51e5d90ef2642 100644 --- a/build.gradle +++ b/build.gradle @@ -320,6 +320,10 @@ subprojects { } logTestStdout.rehydrate(delegate, owner, this)() + // The suites are for running sets of tests in IDEs. + // Gradle will run each test class, so we exclude the suites to avoid redundantly running the tests twice. + exclude '**/*Suite.class' + useJUnit { includeCategories 'org.apache.kafka.test.IntegrationTest' } @@ -338,6 +342,10 @@ subprojects { exceptionFormat = testExceptionFormat } logTestStdout.rehydrate(delegate, owner, this)() + + // The suites are for running sets of tests in IDEs. + // Gradle will run each test class, so we exclude the suites to avoid redundantly running the tests twice. + exclude '**/*Suite.class' if (it.project.name != 'generator') { useJUnit { @@ -457,7 +465,7 @@ subprojects { // See https://www.lightbend.com/blog/scala-inliner-optimizer for more information about the optimizer. scalaCompileOptions.additionalParameters += ["-opt:l:inline"] scalaCompileOptions.additionalParameters += inlineFrom - + // these options are valid for Scala versions < 2.13 only // Scala 2.13 removes them, see https://github.com/scala/scala/pull/6502 and https://github.com/scala/scala/pull/5969 if (versions.baseScala == '2.12') { @@ -1270,11 +1278,26 @@ project(':streams') { standardOutput = new File(generatedDocsDir, "streams_config.html").newOutputStream() } - test { - // The suites are for running sets of tests in IDEs. - // Gradle will run each test class, so we exclude the suites to avoid redundantly running the tests twice. - exclude '**/*Suite.class' - } + task testAll( + dependsOn: [ + ':streams:test', + ':streams:test-utils:test', + ':streams:streams-scala:test', + ':streams:upgrade-system-tests-0100:test', + ':streams:upgrade-system-tests-0101:test', + ':streams:upgrade-system-tests-0102:test', + ':streams:upgrade-system-tests-0110:test', + ':streams:upgrade-system-tests-10:test', + ':streams:upgrade-system-tests-11:test', + ':streams:upgrade-system-tests-20:test', + ':streams:upgrade-system-tests-21:test', + ':streams:upgrade-system-tests-22:test', + ':streams:upgrade-system-tests-23:test', + ':streams:upgrade-system-tests-24:test', + ':streams:upgrade-system-tests-25:test', + ':streams:examples:test' + ] + ) } project(':streams:streams-scala') { From 8ac397bf4ff0007a414629ed86960e79cd1cba9d Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Mon, 28 Sep 2020 19:30:01 -0400 Subject: [PATCH 04/15] KAFKA-10218: Stop reading config topic in every subsequent tick if catchup fails once (#8973) Add logic to reset the existing `canReadConfigs` in `DistributedHerder` once the herder is able to successfully read the configs again. Added unit test to verify the functionality. Author: Chris Egerton Reviewer: Nigel Liang , Randall Hauch --- .../runtime/distributed/DistributedHerder.java | 13 ++++++++++--- .../runtime/distributed/DistributedHerderTest.java | 9 +++++++++ 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java index f1eb4ff1fbec3..d1db3d4902545 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java @@ -313,8 +313,13 @@ public void tick() { try { // if we failed to read to end of log before, we need to make sure the issue was resolved before joining group // Joining and immediately leaving for failure to read configs is exceedingly impolite - if (!canReadConfigs && !readConfigToEnd(workerSyncTimeoutMs)) - return; // Safe to return and tick immediately because readConfigToEnd will do the backoff for us + if (!canReadConfigs) { + if (readConfigToEnd(workerSyncTimeoutMs)) { + canReadConfigs = true; + } else { + return; // Safe to return and tick immediately because readConfigToEnd will do the backoff for us + } + } member.ensureActive(); // Ensure we're in a good state in our group. If not restart and everything should be setup to rejoin @@ -1040,7 +1045,9 @@ private boolean handleRebalanceCompleted() { // we timed out. This should only happen if we failed to read configuration for long enough, // in which case giving back control to the main loop will prevent hanging around indefinitely after getting kicked out of the group. // We also indicate to the main loop that we failed to readConfigs so it will check that the issue was resolved before trying to join the group - if (!readConfigToEnd(workerSyncTimeoutMs)) { + if (readConfigToEnd(workerSyncTimeoutMs)) { + canReadConfigs = true; + } else { canReadConfigs = false; needsRejoin = true; } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java index e0b2b3a820f7e..a1881bbcc164d 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java @@ -1503,6 +1503,11 @@ public void testJoinLeaderCatchUpFails() throws Exception { member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); + // one more tick, to make sure we don't keep trying to read to the config topic unnecessarily + expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + member.poll(EasyMock.anyInt()); + PowerMock.expectLastCall(); + PowerMock.replayAll(); long before = time.milliseconds(); @@ -1520,6 +1525,10 @@ public void testJoinLeaderCatchUpFails() throws Exception { time.sleep(2000L); assertStatistics("leaderUrl", false, 3, 1, 100, 2000L); + // tick once more to ensure that the successful read to the end of the config topic was + // tracked and no further unnecessary attempts were made + herder.tick(); + PowerMock.verifyAll(); } From f716986ac4c35b7a59604fb3370d8d998fd93b6a Mon Sep 17 00:00:00 2001 From: David Arthur Date: Wed, 30 Sep 2020 22:20:20 -0400 Subject: [PATCH 05/15] Backport Jenkinsfile to 2.5 (#9327) --- Jenkinsfile | 145 +++++++++++++++++++++++++++++++++++++++++++++++++++ build.gradle | 4 ++ 2 files changed, 149 insertions(+) create mode 100644 Jenkinsfile diff --git a/Jenkinsfile b/Jenkinsfile new file mode 100644 index 0000000000000..c9f0d5d3b6488 --- /dev/null +++ b/Jenkinsfile @@ -0,0 +1,145 @@ +/* + * + * 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. + * + */ + +def setupGradle() { + // Delete gradle cache to workaround cache corruption bugs, see KAFKA-3167 + dir('.gradle') { + deleteDir() + } + sh './gradlew -version' +} + +def doValidation() { + sh ''' + ./gradlew -PscalaVersion=$SCALA_VERSION clean compileJava compileScala compileTestJava compileTestScala \ + spotlessScalaCheck checkstyleMain checkstyleTest spotbugsMain rat \ + --profile --no-daemon --continue -PxmlSpotBugsReport=true + ''' +} + +def doTest() { + sh ''' + ./gradlew -PscalaVersion=$SCALA_VERSION unitTest integrationTest \ + --profile --no-daemon --continue -PtestLoggingEvents=started,passed,skipped,failed \ + -PignoreFailures=true -PmaxParallelForks=2 -PmaxTestRetries=1 -PmaxTestRetryFailures=5 + ''' + junit '**/build/test-results/**/TEST-*.xml' +} + +def doStreamsArchetype() { + echo 'Verify that Kafka Streams archetype compiles' + + sh ''' + ./gradlew streams:install clients:install connect:json:install connect:api:install \ + || { echo 'Could not install kafka-streams.jar (and dependencies) locally`'; exit 1; } + ''' + + VERSION = sh(script: 'grep "^version=" gradle.properties | cut -d= -f 2', returnStdout: true).trim() + + dir('streams/quickstart') { + sh ''' + mvn clean install -Dgpg.skip \ + || { echo 'Could not `mvn install` streams quickstart archetype'; exit 1; } + ''' + + dir('test-streams-archetype') { + // Note the double quotes for variable interpolation + sh """ + echo "Y" | mvn archetype:generate \ + -DarchetypeCatalog=local \ + -DarchetypeGroupId=org.apache.kafka \ + -DarchetypeArtifactId=streams-quickstart-java \ + -DarchetypeVersion=${VERSION} \ + -DgroupId=streams.examples \ + -DartifactId=streams.examples \ + -Dversion=0.1 \ + -Dpackage=myapps \ + || { echo 'Could not create new project using streams quickstart archetype'; exit 1; } + """ + + dir('streams.examples') { + sh ''' + mvn compile \ + || { echo 'Could not compile streams quickstart archetype project'; exit 1; } + ''' + } + } + } +} + +def tryStreamsArchetype() { + try { + doStreamsArchetype() + } catch(err) { + echo 'Failed to build Kafka Streams archetype, marking this build UNSTABLE' + currentBuild.result = 'UNSTABLE' + } +} + + +pipeline { + agent none + stages { + stage('Build') { + parallel { + stage('JDK 8') { + agent { label 'ubuntu' } + tools { + jdk 'JDK 1.8 (latest)' + maven 'Maven 3.6.3' + } + options { + timeout(time: 8, unit: 'HOURS') + timestamps() + } + environment { + SCALA_VERSION=2.12 + } + steps { + setupGradle() + doValidation() + doTest() + tryStreamsArchetype() + } + } + + stage('JDK 11') { + agent { label 'ubuntu' } + tools { + jdk 'JDK 11 (latest)' + } + options { + timeout(time: 8, unit: 'HOURS') + timestamps() + } + environment { + SCALA_VERSION=2.13 + } + steps { + setupGradle() + doValidation() + doTest() + echo 'Skipping Kafka Streams archetype test for Java 11' + } + } + } + } + } +} + diff --git a/build.gradle b/build.gradle index 51e5d90ef2642..c7bb31795a1f0 100644 --- a/build.gradle +++ b/build.gradle @@ -112,6 +112,7 @@ ext { buildVersionFileName = "kafka-version.properties" userMaxForks = project.hasProperty('maxParallelForks') ? maxParallelForks.toInteger() : null + userIgnoreFailures = project.hasProperty('ignoreFailures') ? ignoreFailures : false skipSigning = project.hasProperty('skipSigning') && skipSigning.toBoolean() shouldSign = !skipSigning && !version.endsWith("SNAPSHOT") && project.gradle.startParameter.taskNames.any { it.contains("upload") } @@ -295,6 +296,7 @@ subprojects { test { maxParallelForks = userMaxForks ?: Runtime.runtime.availableProcessors() + ignoreFailures = userIgnoreFailures minHeapSize = "256m" maxHeapSize = "2048m" @@ -309,6 +311,7 @@ subprojects { task integrationTest(type: Test, dependsOn: compileJava) { maxParallelForks = userMaxForks ?: Runtime.runtime.availableProcessors() + ignoreFailures = userIgnoreFailures minHeapSize = "256m" maxHeapSize = "2048m" @@ -332,6 +335,7 @@ subprojects { task unitTest(type: Test, dependsOn: compileJava) { maxParallelForks = userMaxForks ?: Runtime.runtime.availableProcessors() + ignoreFailures = userIgnoreFailures minHeapSize = "256m" maxHeapSize = "2048m" From 94fd8ff345047043eca4f9ed10116a5f41af7bf0 Mon Sep 17 00:00:00 2001 From: Shaik Zakir Hussain Date: Fri, 2 Oct 2020 19:51:11 +0530 Subject: [PATCH 06/15] KAFKA-10477: Fix JsonConverter regression to treat MISSING nodes as NULL nodes (#9306) Fixes a regression introduced in `JsonConverter` with previous upgrades from Jackson Databind 2.9.x to 2.10.x. Jackson Databind version 2.10.0 included a backward-incompatible behavioral change to use `JsonNodeType.MISSING` (and `MissingNode`, the subclass of `JsonNode` that has a type of `MISSING`) instead of `JsonNodeType.NULL` / `NullNode`. See https://github.com/FasterXML/jackson-databind/issues/2211 for details of this change. This change makes recovers the older `JsonConverter` behavior of returning null on empty input. Added two unit tests for this change. Both unit tests were independently tested with earlier released versions and passed on all versions that used Jackson 2.9.x and earlier, and failed on all versions that used 2.10.x and that did not have the fixed included in the PR. Both of the new unit tests pass with this fix to `JsonConverter`. Author: Shaik Zakir Hussain Reviewer: Randall Hauch --- .../kafka/connect/json/JsonConverter.java | 2 +- .../kafka/connect/json/JsonConverterTest.java | 30 +++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java b/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java index 8a2d6768cc96f..1a175381e8f11 100644 --- a/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java +++ b/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java @@ -729,6 +729,7 @@ private static Object convertToConnect(Schema schema, JsonNode jsonValue) { } else { switch (jsonValue.getNodeType()) { case NULL: + case MISSING: // Special case. With no schema return null; case BOOLEAN: @@ -751,7 +752,6 @@ private static Object convertToConnect(Schema schema, JsonNode jsonValue) { break; case BINARY: - case MISSING: case POJO: default: schemaType = null; diff --git a/connect/json/src/test/java/org/apache/kafka/connect/json/JsonConverterTest.java b/connect/json/src/test/java/org/apache/kafka/connect/json/JsonConverterTest.java index 2e189e2d584ae..a1ac71d4c8d57 100644 --- a/connect/json/src/test/java/org/apache/kafka/connect/json/JsonConverterTest.java +++ b/connect/json/src/test/java/org/apache/kafka/connect/json/JsonConverterTest.java @@ -195,6 +195,36 @@ public void nullToConnect() { assertEquals(SchemaAndValue.NULL, converted); } + /** + * When schemas are disabled, empty data should be decoded to an empty envelope. + * This test verifies the case where `schemas.enable` configuration is set to false, and + * {@link JsonConverter} converts empty bytes to {@link SchemaAndValue#NULL}. + */ + @Test + public void emptyBytesToConnect() { + // This characterizes the messages with empty data when Json schemas is disabled + Map props = Collections.singletonMap("schemas.enable", false); + converter.configure(props, true); + SchemaAndValue converted = converter.toConnectData(TOPIC, "".getBytes()); + assertEquals(SchemaAndValue.NULL, converted); + } + + /** + * When schemas are disabled, fields are mapped to Connect maps. + */ + @Test + public void schemalessWithEmptyFieldValueToConnect() { + // This characterizes the messages with empty data when Json schemas is disabled + Map props = Collections.singletonMap("schemas.enable", false); + converter.configure(props, true); + String input = "{ \"a\": \"\", \"b\": null}"; + SchemaAndValue converted = converter.toConnectData(TOPIC, input.getBytes()); + Map expected = new HashMap<>(); + expected.put("a", ""); + expected.put("b", null); + assertEquals(new SchemaAndValue(null, expected), converted); + } + @Test public void nullSchemaPrimitiveToConnect() { SchemaAndValue converted = converter.toConnectData(TOPIC, "{ \"schema\": null, \"payload\": null }".getBytes()); From d2c332a1bebb02f876bbb6e3afeb75f952b8142f Mon Sep 17 00:00:00 2001 From: Vikas Singh Date: Mon, 5 Oct 2020 12:06:55 -0700 Subject: [PATCH 07/15] KAFKA-10531: Check for negative values to Thread.sleep call (#9347) System.currentTimeMillis() is not monotonic, so using that to calculate time to sleep can result in negative values. That will throw IllegalArgumentException. This change checks for that and sleeps for a second (to avoid tight loop) if the value returned is negative. Author: Shaik Zakir Hussain Reviewer: Randall Hauch --- .../apache/kafka/connect/util/KafkaBasedLog.java | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java b/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java index 69d2588fdea7a..5248715aa6293 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java @@ -31,7 +31,6 @@ import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.utils.Time; -import org.apache.kafka.common.utils.Utils; import org.apache.kafka.connect.errors.ConnectException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -45,6 +44,7 @@ import java.util.Queue; import java.util.Set; import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; /** @@ -70,7 +70,8 @@ */ public class KafkaBasedLog { private static final Logger log = LoggerFactory.getLogger(KafkaBasedLog.class); - private static final long CREATE_TOPIC_TIMEOUT_MS = 30000; + private static final long CREATE_TOPIC_TIMEOUT_NS = TimeUnit.SECONDS.toNanos(30); + private static final long MAX_SLEEP_MS = TimeUnit.SECONDS.toMillis(1); private Time time; private final String topic; @@ -133,11 +134,13 @@ public void start() { List partitions = new ArrayList<>(); // We expect that the topics will have been created either manually by the user or automatically by the herder - List partitionInfos = null; - long started = time.milliseconds(); - while (partitionInfos == null && time.milliseconds() - started < CREATE_TOPIC_TIMEOUT_MS) { + List partitionInfos = consumer.partitionsFor(topic); + long started = time.nanoseconds(); + long sleepMs = 100; + while (partitionInfos == null && time.nanoseconds() - started < CREATE_TOPIC_TIMEOUT_NS) { + time.sleep(sleepMs); + sleepMs = Math.min(2 * sleepMs, MAX_SLEEP_MS); partitionInfos = consumer.partitionsFor(topic); - Utils.sleep(Math.min(time.milliseconds() - started, 1000)); } if (partitionInfos == null) throw new ConnectException("Could not look up partition metadata for offset backing store topic in" + From 25a1ed4cb5975f7531e3c65e5f5d8ab77f5f3a4b Mon Sep 17 00:00:00 2001 From: Boyang Chen Date: Tue, 14 Apr 2020 09:49:03 -0700 Subject: [PATCH 08/15] MINOR: remove stream simple benchmark suite (#8353) Reviewers: Matthias J. Sax --- .../kafka/streams/perf/SimpleBenchmark.java | 752 ------------------ .../kafka/streams/perf/YahooBenchmark.java | 306 ------- .../kafkatest/benchmarks/streams/__init__.py | 14 - .../streams/streams_simple_benchmark_test.py | 164 ---- .../performance/streams_performance.py | 108 --- 5 files changed, 1344 deletions(-) delete mode 100644 streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java delete mode 100644 streams/src/test/java/org/apache/kafka/streams/perf/YahooBenchmark.java delete mode 100644 tests/kafkatest/benchmarks/streams/__init__.py delete mode 100644 tests/kafkatest/benchmarks/streams/streams_simple_benchmark_test.py delete mode 100644 tests/kafkatest/services/performance/streams_performance.py diff --git a/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java b/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java deleted file mode 100644 index 4a14b87695e62..0000000000000 --- a/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java +++ /dev/null @@ -1,752 +0,0 @@ -/* - * 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.streams.perf; - -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.clients.producer.ProducerConfig; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.PartitionInfo; -import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.serialization.ByteArrayDeserializer; -import org.apache.kafka.common.serialization.ByteArraySerializer; -import org.apache.kafka.common.serialization.IntegerDeserializer; -import org.apache.kafka.common.serialization.IntegerSerializer; -import org.apache.kafka.common.serialization.Serde; -import org.apache.kafka.common.serialization.Serdes; -import org.apache.kafka.common.utils.Utils; -import org.apache.kafka.streams.KafkaStreams; -import org.apache.kafka.streams.StreamsBuilder; -import org.apache.kafka.streams.StreamsConfig; -import org.apache.kafka.streams.kstream.Consumed; -import org.apache.kafka.streams.kstream.ForeachAction; -import org.apache.kafka.streams.kstream.JoinWindows; -import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.kstream.KTable; -import org.apache.kafka.streams.kstream.TimeWindows; -import org.apache.kafka.streams.kstream.ValueJoiner; -import org.apache.kafka.streams.kstream.Windowed; -import org.apache.kafka.streams.processor.AbstractProcessor; -import org.apache.kafka.streams.processor.Processor; -import org.apache.kafka.streams.processor.ProcessorContext; -import org.apache.kafka.streams.processor.ProcessorSupplier; -import org.apache.kafka.streams.state.KeyValueIterator; -import org.apache.kafka.streams.state.KeyValueStore; -import org.apache.kafka.streams.state.StoreBuilder; -import org.apache.kafka.streams.state.Stores; -import org.apache.kafka.streams.state.WindowStore; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; -import java.util.Locale; -import java.util.Properties; -import java.util.Random; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - -import static java.time.Duration.ofMillis; -import static java.time.Duration.ofSeconds; -import static java.time.Instant.ofEpochMilli; - -/** - * Class that provides support for a series of benchmarks. It is usually driven by - * tests/kafkatest/benchmarks/streams/streams_simple_benchmark_test.py. - * If ran manually through the main() function below, you must do the following: - * 1. Have ZK and a Kafka broker set up - * 2. Run the loading step first: SimpleBenchmark localhost:9092 /tmp/statedir numRecords true "all" - * 3. Run the stream processing step second: SimpleBenchmark localhost:9092 /tmp/statedir numRecords false "all" - * Note that what changed is the 4th parameter, from "true" indicating that is a load phase, to "false" indicating - * that this is a real run. - * - * Note that "all" is a convenience option when running this test locally and will not work when running the test - * at scale (through tests/kafkatest/benchmarks/streams/streams_simple_benchmark_test.py). That is due to exact syncronization - * needs for each test (e.g., you wouldn't want one instance to run "count" while another - * is still running "consume" - */ -public class SimpleBenchmark { - private static final String LOADING_PRODUCER_CLIENT_ID = "simple-benchmark-loading-producer"; - - private static final String SOURCE_TOPIC_ONE = "simpleBenchmarkSourceTopic1"; - private static final String SOURCE_TOPIC_TWO = "simpleBenchmarkSourceTopic2"; - private static final String SINK_TOPIC = "simpleBenchmarkSinkTopic"; - - private static final String YAHOO_CAMPAIGNS_TOPIC = "yahooCampaigns"; - private static final String YAHOO_EVENTS_TOPIC = "yahooEvents"; - - private static final ValueJoiner VALUE_JOINER = new ValueJoiner() { - @Override - public byte[] apply(final byte[] value1, final byte[] value2) { - // dump joiner in order to have as less join overhead as possible - if (value1 != null) { - return value1; - } else if (value2 != null) { - return value2; - } else { - return new byte[100]; - } - } - }; - - private static final Serde BYTE_SERDE = Serdes.ByteArray(); - private static final Serde INTEGER_SERDE = Serdes.Integer(); - - long processedBytes = 0L; - int processedRecords = 0; - - private static final long POLL_MS = 500L; - private static final long COMMIT_INTERVAL_MS = 30000L; - private static final int MAX_POLL_RECORDS = 1000; - - /* ----------- benchmark variables that are hard-coded ----------- */ - - private static final int KEY_SPACE_SIZE = 10000; - - private static final long STREAM_STREAM_JOIN_WINDOW = 10000L; - - private static final long AGGREGATE_WINDOW_SIZE = 1000L; - - private static final long AGGREGATE_WINDOW_ADVANCE = 500L; - - private static final int SOCKET_SIZE_BYTES = 1024 * 1024; - - // the following numbers are based on empirical results and should only - // be considered for updates when perf results have significantly changed - - // with at least 10 million records, we run for at most 3 minutes - private static final int MAX_WAIT_MS = 3 * 60 * 1000; - - /* ----------- benchmark variables that can be specified ----------- */ - - final String testName; - - final int numRecords; - - final Properties props; - - private final int valueSize; - - private final double keySkew; - - /* ----------- ----------------------------------------- ----------- */ - - - private SimpleBenchmark(final Properties props, - final String testName, - final int numRecords, - final double keySkew, - final int valueSize) { - super(); - this.props = props; - this.testName = testName; - this.keySkew = keySkew; - this.valueSize = valueSize; - this.numRecords = numRecords; - } - - private void run() { - switch (testName) { - // loading phases - case "load-one": - produce(LOADING_PRODUCER_CLIENT_ID, SOURCE_TOPIC_ONE, numRecords, keySkew, valueSize); - break; - case "load-two": - produce(LOADING_PRODUCER_CLIENT_ID, SOURCE_TOPIC_ONE, numRecords, keySkew, valueSize); - produce(LOADING_PRODUCER_CLIENT_ID, SOURCE_TOPIC_TWO, numRecords, keySkew, valueSize); - break; - - // testing phases - case "consume": - consume(SOURCE_TOPIC_ONE); - break; - case "consumeproduce": - consumeAndProduce(SOURCE_TOPIC_ONE); - break; - case "streamcount": - countStreamsNonWindowed(SOURCE_TOPIC_ONE); - break; - case "streamcountwindowed": - countStreamsWindowed(SOURCE_TOPIC_ONE); - break; - case "streamprocess": - processStream(SOURCE_TOPIC_ONE); - break; - case "streamprocesswithsink": - processStreamWithSink(SOURCE_TOPIC_ONE); - break; - case "streamprocesswithstatestore": - processStreamWithStateStore(SOURCE_TOPIC_ONE); - break; - case "streamprocesswithwindowstore": - processStreamWithWindowStore(SOURCE_TOPIC_ONE); - break; - case "streamtablejoin": - streamTableJoin(SOURCE_TOPIC_ONE, SOURCE_TOPIC_TWO); - break; - case "streamstreamjoin": - streamStreamJoin(SOURCE_TOPIC_ONE, SOURCE_TOPIC_TWO); - break; - case "tabletablejoin": - tableTableJoin(SOURCE_TOPIC_ONE, SOURCE_TOPIC_TWO); - break; - case "yahoo": - yahooBenchmark(YAHOO_CAMPAIGNS_TOPIC, YAHOO_EVENTS_TOPIC); - break; - default: - throw new RuntimeException("Unknown test name " + testName); - - } - } - - public static void main(final String[] args) throws IOException { - if (args.length < 5) { - System.err.println("Not enough parameters are provided; expecting propFileName, testName, numRecords, keySkew, valueSize"); - System.exit(1); - } - - final String propFileName = args[0]; - final String testName = args[1].toLowerCase(Locale.ROOT); - final int numRecords = Integer.parseInt(args[2]); - final double keySkew = Double.parseDouble(args[3]); // 0d means even distribution - final int valueSize = Integer.parseInt(args[4]); - - final Properties props = Utils.loadProps(propFileName); - final String kafka = props.getProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG); - - if (kafka == null) { - System.err.println("No bootstrap kafka servers specified in " + StreamsConfig.BOOTSTRAP_SERVERS_CONFIG); - System.exit(1); - } - - // Note: this output is needed for automated tests and must not be removed - System.out.println("StreamsTest instance started"); - - System.out.println("testName=" + testName); - System.out.println("streamsProperties=" + props); - System.out.println("numRecords=" + numRecords); - System.out.println("keySkew=" + keySkew); - System.out.println("valueSize=" + valueSize); - - final SimpleBenchmark benchmark = new SimpleBenchmark(props, testName, numRecords, keySkew, valueSize); - - benchmark.run(); - } - - public void setStreamProperties(final String applicationId) { - props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId); - props.put(StreamsConfig.CLIENT_ID_CONFIG, "simple-benchmark"); - props.put(StreamsConfig.POLL_MS_CONFIG, POLL_MS); - props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, COMMIT_INTERVAL_MS); - props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass()); - props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArray().getClass()); - // the socket buffer needs to be large, especially when running in AWS with - // high latency. if running locally the default is fine. - props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - props.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG, SOCKET_SIZE_BYTES); - props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, MAX_POLL_RECORDS); - - // improve producer throughput - props.put(ProducerConfig.LINGER_MS_CONFIG, 5000); - props.put(ProducerConfig.BATCH_SIZE_CONFIG, 128 * 1024); - } - - private Properties setProduceConsumeProperties(final String clientId) { - final Properties clientProps = new Properties(); - clientProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, props.getProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG)); - clientProps.put(ProducerConfig.CLIENT_ID_CONFIG, clientId); - // the socket buffer needs to be large, especially when running in AWS with - // high latency. if running locally the default is fine. - clientProps.put(ProducerConfig.LINGER_MS_CONFIG, 5000); - clientProps.put(ProducerConfig.BATCH_SIZE_CONFIG, 128 * 1024); - clientProps.put(ProducerConfig.SEND_BUFFER_CONFIG, SOCKET_SIZE_BYTES); - clientProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class); - clientProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); - clientProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class); - clientProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); - clientProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); - // the socket buffer needs to be large, especially when running in AWS with - // high latency. if running locally the default is fine. - clientProps.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG, SOCKET_SIZE_BYTES); - clientProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, MAX_POLL_RECORDS); - return clientProps; - } - - void resetStats() { - processedRecords = 0; - processedBytes = 0L; - } - - /** - * Produce values to a topic - * @param clientId String specifying client ID - * @param topic Topic to produce to - * @param numRecords Number of records to produce - * @param keySkew Key zipf distribution skewness - * @param valueSize Size of value in bytes - */ - private void produce(final String clientId, - final String topic, - final int numRecords, - final double keySkew, - final int valueSize) { - final Properties props = setProduceConsumeProperties(clientId); - final ZipfGenerator keyGen = new ZipfGenerator(KEY_SPACE_SIZE, keySkew); - - try (final KafkaProducer producer = new KafkaProducer<>(props)) { - final byte[] value = new byte[valueSize]; - // put some random values to increase entropy. Some devices - // like SSDs do compression and if the array is all zeros - // the performance will be too good. - new Random(System.currentTimeMillis()).nextBytes(value); - - for (int i = 0; i < numRecords; i++) { - producer.send(new ProducerRecord<>(topic, keyGen.next(), value)); - } - } - } - - private void consumeAndProduce(final String topic) { - final Properties consumerProps = setProduceConsumeProperties("simple-benchmark-consumer"); - final Properties producerProps = setProduceConsumeProperties("simple-benchmark-producer"); - - final long startTime = System.currentTimeMillis(); - try (final KafkaConsumer consumer = new KafkaConsumer<>(consumerProps); - final KafkaProducer producer = new KafkaProducer<>(producerProps)) { - final List partitions = getAllPartitions(consumer, topic); - - consumer.assign(partitions); - consumer.seekToBeginning(partitions); - - while (true) { - final ConsumerRecords records = consumer.poll(ofMillis(POLL_MS)); - if (records.isEmpty()) { - if (processedRecords == numRecords) { - break; - } - } else { - for (final ConsumerRecord record : records) { - producer.send(new ProducerRecord<>(SINK_TOPIC, record.key(), record.value())); - processedRecords++; - processedBytes += record.value().length + Integer.SIZE; - if (processedRecords == numRecords) { - break; - } - } - } - if (processedRecords == numRecords) { - break; - } - } - } - - final long endTime = System.currentTimeMillis(); - - printResults("ConsumerProducer Performance [records/latency/rec-sec/MB-sec read]: ", endTime - startTime); - } - - private void consume(final String topic) { - final Properties consumerProps = setProduceConsumeProperties("simple-benchmark-consumer"); - - final long startTime = System.currentTimeMillis(); - - try (final KafkaConsumer consumer = new KafkaConsumer<>(consumerProps)) { - final List partitions = getAllPartitions(consumer, topic); - - consumer.assign(partitions); - consumer.seekToBeginning(partitions); - - while (true) { - final ConsumerRecords records = consumer.poll(ofMillis(POLL_MS)); - if (records.isEmpty()) { - if (processedRecords == numRecords) { - break; - } - } else { - for (final ConsumerRecord record : records) { - processedRecords++; - processedBytes += record.value().length + Integer.SIZE; - if (processedRecords == numRecords) { - break; - } - } - } - if (processedRecords == numRecords) { - break; - } - } - } - - final long endTime = System.currentTimeMillis(); - - printResults("Consumer Performance [records/latency/rec-sec/MB-sec read]: ", endTime - startTime); - } - - private void processStream(final String topic) { - final CountDownLatch latch = new CountDownLatch(1); - - setStreamProperties("simple-benchmark-streams-source"); - - final StreamsBuilder builder = new StreamsBuilder(); - - builder.stream(topic, Consumed.with(INTEGER_SERDE, BYTE_SERDE)).peek(new CountDownAction(latch)); - - final KafkaStreams streams = createKafkaStreamsWithExceptionHandler(builder, props); - runGenericBenchmark(streams, "Streams Source Performance [records/latency/rec-sec/MB-sec joined]: ", latch); - } - - private void processStreamWithSink(final String topic) { - final CountDownLatch latch = new CountDownLatch(1); - - setStreamProperties("simple-benchmark-streams-source-sink"); - - final StreamsBuilder builder = new StreamsBuilder(); - - final KStream source = builder.stream(topic); - source.peek(new CountDownAction(latch)).to(SINK_TOPIC); - - final KafkaStreams streams = createKafkaStreamsWithExceptionHandler(builder, props); - runGenericBenchmark(streams, "Streams SourceSink Performance [records/latency/rec-sec/MB-sec joined]: ", latch); - } - - private void processStreamWithStateStore(final String topic) { - final CountDownLatch latch = new CountDownLatch(1); - - setStreamProperties("simple-benchmark-streams-with-store"); - - final StreamsBuilder builder = new StreamsBuilder(); - final StoreBuilder> storeBuilder = - Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("store"), INTEGER_SERDE, BYTE_SERDE); - builder.addStateStore(storeBuilder.withCachingEnabled()); - - final KStream source = builder.stream(topic); - - source.peek(new CountDownAction(latch)).process(new ProcessorSupplier() { - @Override - public Processor get() { - return new AbstractProcessor() { - KeyValueStore store; - - @SuppressWarnings("unchecked") - @Override - public void init(final ProcessorContext context) { - super.init(context); - store = (KeyValueStore) context.getStateStore("store"); - } - - @Override - public void process(final Integer key, final byte[] value) { - store.get(key); - store.put(key, value); - } - }; - } - }, "store"); - - final KafkaStreams streams = createKafkaStreamsWithExceptionHandler(builder, props); - runGenericBenchmark(streams, "Streams Stateful Performance [records/latency/rec-sec/MB-sec joined]: ", latch); - } - - private void processStreamWithWindowStore(final String topic) { - final CountDownLatch latch = new CountDownLatch(1); - - setStreamProperties("simple-benchmark-streams-with-store"); - - final StreamsBuilder builder = new StreamsBuilder(); - - final StoreBuilder> storeBuilder = Stores.windowStoreBuilder( - Stores.persistentWindowStore( - "store", - ofMillis(AGGREGATE_WINDOW_SIZE * 3), - ofMillis(AGGREGATE_WINDOW_SIZE), - false - ), - INTEGER_SERDE, - BYTE_SERDE - ); - builder.addStateStore(storeBuilder.withCachingEnabled()); - - final KStream source = builder.stream(topic); - - source.peek(new CountDownAction(latch)).process(new ProcessorSupplier() { - @Override - public Processor get() { - return new AbstractProcessor() { - WindowStore store; - - @SuppressWarnings("unchecked") - @Override - public void init(final ProcessorContext context) { - super.init(context); - store = (WindowStore) context.getStateStore("store"); - } - - @Override - public void process(final Integer key, final byte[] value) { - final long timestamp = context().timestamp(); - final KeyValueIterator, byte[]> iter = store.fetch(key - 10, key + 10, ofEpochMilli(timestamp - 1000L), ofEpochMilli(timestamp)); - while (iter.hasNext()) { - iter.next(); - } - iter.close(); - - store.put(key, value, timestamp); - } - }; - } - }, "store"); - - final KafkaStreams streams = createKafkaStreamsWithExceptionHandler(builder, props); - runGenericBenchmark(streams, "Streams Stateful Performance [records/latency/rec-sec/MB-sec joined]: ", latch); - } - - /** - * Measure the performance of a simple aggregate like count. - * Counts the occurrence of numbers (note that normally people count words, this - * example counts numbers) - */ - private void countStreamsNonWindowed(final String sourceTopic) { - final CountDownLatch latch = new CountDownLatch(1); - - setStreamProperties("simple-benchmark-nonwindowed-count"); - - final StreamsBuilder builder = new StreamsBuilder(); - final KStream input = builder.stream(sourceTopic); - - input.peek(new CountDownAction(latch)) - .groupByKey() - .count(); - - final KafkaStreams streams = createKafkaStreamsWithExceptionHandler(builder, props); - runGenericBenchmark(streams, "Streams Count Performance [records/latency/rec-sec/MB-sec counted]: ", latch); - } - - /** - * Measure the performance of a simple aggregate like count. - * Counts the occurrence of numbers (note that normally people count words, this - * example counts numbers) - */ - private void countStreamsWindowed(final String sourceTopic) { - final CountDownLatch latch = new CountDownLatch(1); - - setStreamProperties("simple-benchmark-windowed-count"); - - final StreamsBuilder builder = new StreamsBuilder(); - final KStream input = builder.stream(sourceTopic); - - input.peek(new CountDownAction(latch)) - .groupByKey() - .windowedBy(TimeWindows.of(ofMillis(AGGREGATE_WINDOW_SIZE)).advanceBy(ofMillis(AGGREGATE_WINDOW_ADVANCE))) - .count(); - - final KafkaStreams streams = createKafkaStreamsWithExceptionHandler(builder, props); - runGenericBenchmark(streams, "Streams Count Windowed Performance [records/latency/rec-sec/MB-sec counted]: ", latch); - } - - /** - * Measure the performance of a KStream-KTable left join. The setup is such that each - * KStream record joins to exactly one element in the KTable - */ - private void streamTableJoin(final String kStreamTopic, final String kTableTopic) { - final CountDownLatch latch = new CountDownLatch(1); - - setStreamProperties("simple-benchmark-stream-table-join"); - - final StreamsBuilder builder = new StreamsBuilder(); - - final KStream input1 = builder.stream(kStreamTopic); - final KTable input2 = builder.table(kTableTopic); - - input1.leftJoin(input2, VALUE_JOINER).foreach(new CountDownAction(latch)); - - final KafkaStreams streams = createKafkaStreamsWithExceptionHandler(builder, props); - - // run benchmark - runGenericBenchmark(streams, "Streams KStreamKTable LeftJoin Performance [records/latency/rec-sec/MB-sec joined]: ", latch); - } - - /** - * Measure the performance of a KStream-KStream left join. The setup is such that each - * KStream record joins to exactly one element in the other KStream - */ - private void streamStreamJoin(final String kStreamTopic1, final String kStreamTopic2) { - final CountDownLatch latch = new CountDownLatch(1); - - setStreamProperties("simple-benchmark-stream-stream-join"); - - final StreamsBuilder builder = new StreamsBuilder(); - - final KStream input1 = builder.stream(kStreamTopic1); - final KStream input2 = builder.stream(kStreamTopic2); - - input1.leftJoin(input2, VALUE_JOINER, JoinWindows.of(ofMillis(STREAM_STREAM_JOIN_WINDOW))).foreach(new CountDownAction(latch)); - - final KafkaStreams streams = createKafkaStreamsWithExceptionHandler(builder, props); - - // run benchmark - runGenericBenchmark(streams, "Streams KStreamKStream LeftJoin Performance [records/latency/rec-sec/MB-sec joined]: ", latch); - } - - /** - * Measure the performance of a KTable-KTable left join. The setup is such that each - * KTable record joins to exactly one element in the other KTable - */ - private void tableTableJoin(final String kTableTopic1, final String kTableTopic2) { - final CountDownLatch latch = new CountDownLatch(1); - - // setup join - setStreamProperties("simple-benchmark-table-table-join"); - - final StreamsBuilder builder = new StreamsBuilder(); - - final KTable input1 = builder.table(kTableTopic1); - final KTable input2 = builder.table(kTableTopic2); - - input1.leftJoin(input2, VALUE_JOINER).toStream().foreach(new CountDownAction(latch)); - - final KafkaStreams streams = createKafkaStreamsWithExceptionHandler(builder, props); - - // run benchmark - runGenericBenchmark(streams, "Streams KTableKTable LeftJoin Performance [records/latency/rec-sec/MB-sec joined]: ", latch); - } - - void printResults(final String nameOfBenchmark, final long latency) { - System.out.println(nameOfBenchmark + - processedRecords + "/" + - latency + "/" + - recordsPerSec(latency, processedRecords) + "/" + - megabytesPerSec(latency, processedBytes)); - } - - void runGenericBenchmark(final KafkaStreams streams, final String nameOfBenchmark, final CountDownLatch latch) { - streams.start(); - - final long startTime = System.currentTimeMillis(); - long endTime = startTime; - - while (latch.getCount() > 0 && (endTime - startTime < MAX_WAIT_MS)) { - try { - latch.await(1000, TimeUnit.MILLISECONDS); - } catch (final InterruptedException ex) { - Thread.interrupted(); - } - - endTime = System.currentTimeMillis(); - } - streams.close(); - - printResults(nameOfBenchmark, endTime - startTime); - } - - private class CountDownAction implements ForeachAction { - private final CountDownLatch latch; - - CountDownAction(final CountDownLatch latch) { - this.latch = latch; - } - - @Override - public void apply(final Integer key, final byte[] value) { - processedRecords++; - processedBytes += Integer.SIZE + value.length; - - if (processedRecords == numRecords) { - this.latch.countDown(); - } - } - } - - private KafkaStreams createKafkaStreamsWithExceptionHandler(final StreamsBuilder builder, final Properties props) { - final KafkaStreams streamsClient = new KafkaStreams(builder.build(), props); - streamsClient.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { - @Override - public void uncaughtException(final Thread t, final Throwable e) { - System.out.println("FATAL: An unexpected exception is encountered on thread " + t + ": " + e); - - streamsClient.close(ofSeconds(30)); - } - }); - - return streamsClient; - } - - private double megabytesPerSec(final long time, final long processedBytes) { - return (processedBytes / 1024.0 / 1024.0) / (time / 1000.0); - } - - private double recordsPerSec(final long time, final int numRecords) { - return numRecords / (time / 1000.0); - } - - private List getAllPartitions(final KafkaConsumer consumer, final String... topics) { - final ArrayList partitions = new ArrayList<>(); - - for (final String topic : topics) { - for (final PartitionInfo info : consumer.partitionsFor(topic)) { - partitions.add(new TopicPartition(info.topic(), info.partition())); - } - } - return partitions; - } - - private void yahooBenchmark(final String campaignsTopic, final String eventsTopic) { - final YahooBenchmark benchmark = new YahooBenchmark(this, campaignsTopic, eventsTopic); - - benchmark.run(); - } - - private class ZipfGenerator { - final private Random rand = new Random(System.currentTimeMillis()); - final private int size; - final private double skew; - - private double bottom = 0.0d; - - ZipfGenerator(final int size, final double skew) { - this.size = size; - this.skew = skew; - - for (int i = 1; i < size; i++) { - this.bottom += 1.0d / Math.pow(i, this.skew); - } - } - - int next() { - if (skew == 0.0d) { - return rand.nextInt(size); - } else { - int rank; - double dice; - double frequency; - - rank = rand.nextInt(size); - frequency = (1.0d / Math.pow(rank, this.skew)) / this.bottom; - dice = rand.nextDouble(); - - while (!(dice < frequency)) { - rank = rand.nextInt(size); - frequency = (1.0d / Math.pow(rank, this.skew)) / this.bottom; - dice = rand.nextDouble(); - } - - return rank; - } - } - } -} diff --git a/streams/src/test/java/org/apache/kafka/streams/perf/YahooBenchmark.java b/streams/src/test/java/org/apache/kafka/streams/perf/YahooBenchmark.java deleted file mode 100644 index 2cab626417bd7..0000000000000 --- a/streams/src/test/java/org/apache/kafka/streams/perf/YahooBenchmark.java +++ /dev/null @@ -1,306 +0,0 @@ -/* - * 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.streams.perf; - -import com.fasterxml.jackson.databind.ObjectMapper; -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.clients.producer.ProducerConfig; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.errors.SerializationException; -import org.apache.kafka.common.serialization.ByteArraySerializer; -import org.apache.kafka.common.serialization.Deserializer; -import org.apache.kafka.common.serialization.Serdes; -import org.apache.kafka.common.serialization.Serializer; -import org.apache.kafka.common.serialization.StringSerializer; -import org.apache.kafka.streams.KafkaStreams; -import org.apache.kafka.streams.StreamsBuilder; -import org.apache.kafka.streams.StreamsConfig; -import org.apache.kafka.streams.kstream.Consumed; -import org.apache.kafka.streams.kstream.Grouped; -import org.apache.kafka.streams.kstream.Joined; -import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.kstream.KTable; -import org.apache.kafka.streams.kstream.Materialized; -import org.apache.kafka.streams.kstream.TimeWindows; - -import java.time.Duration; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import java.util.Random; -import java.util.UUID; -import java.util.concurrent.CountDownLatch; - - -/** - * A basic DSL and data generation that emulates the behavior of the Yahoo Benchmark - * https://yahooeng.tumblr.com/post/135321837876/benchmarking-streaming-computation-engines-at - * Thanks to Michael Armbrust for providing the initial code for this benchmark in his blog: - * https://databricks.com/blog/2017/06/06/simple-super-fast-streaming-engine-apache-spark.html - */ -public class YahooBenchmark { - private final SimpleBenchmark parent; - private final String campaignsTopic; - private final String eventsTopic; - - static class ProjectedEvent { - /* attributes need to be public for serializer to work */ - /* main attributes */ - String eventType; - String adID; - - /* other attributes */ - long eventTime; - /* not used - public String userID = UUID.randomUUID().toString(); - public String pageID = UUID.randomUUID().toString(); - public String addType = "banner78"; - public String ipAddress = "1.2.3.4"; - */ - } - - static class CampaignAd { - /* attributes need to be public for serializer to work */ - String adID; - String campaignID; - } - - @SuppressWarnings("WeakerAccess") - public YahooBenchmark(final SimpleBenchmark parent, final String campaignsTopic, final String eventsTopic) { - this.parent = parent; - this.campaignsTopic = campaignsTopic; - this.eventsTopic = eventsTopic; - } - - // just for Yahoo benchmark - private boolean maybeSetupPhaseCampaigns(final String topic, - final String clientId, - final boolean skipIfAllTests, - final int numCampaigns, - final int adsPerCampaign, - final List ads) { - parent.resetStats(); - // initialize topics - System.out.println("Initializing topic " + topic); - - final Properties props = new Properties(); - props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, parent.props.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG)); - props.put(ProducerConfig.CLIENT_ID_CONFIG, clientId); - props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); - props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); - - try (final KafkaProducer producer = new KafkaProducer<>(props)) { - for (int c = 0; c < numCampaigns; c++) { - final String campaignID = UUID.randomUUID().toString(); - for (int a = 0; a < adsPerCampaign; a++) { - final String adId = UUID.randomUUID().toString(); - final String concat = adId + ":" + campaignID; - producer.send(new ProducerRecord<>(topic, adId, concat)); - ads.add(adId); - parent.processedRecords++; - parent.processedBytes += concat.length() + adId.length(); - } - } - } - return true; - } - - // just for Yahoo benchmark - private void maybeSetupPhaseEvents(final String topic, - final String clientId, - final int numRecords, - final List ads) { - parent.resetStats(); - final String[] eventTypes = new String[]{"view", "click", "purchase"}; - final Random rand = new Random(System.currentTimeMillis()); - System.out.println("Initializing topic " + topic); - - final Properties props = new Properties(); - props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, parent.props.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG)); - props.put(ProducerConfig.CLIENT_ID_CONFIG, clientId); - props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); - props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); - - final long startTime = System.currentTimeMillis(); - - try (final KafkaProducer producer = new KafkaProducer<>(props)) { - final ProjectedEvent event = new ProjectedEvent(); - final Map serdeProps = new HashMap<>(); - final Serializer projectedEventSerializer = new JsonPOJOSerializer<>(); - serdeProps.put("JsonPOJOClass", ProjectedEvent.class); - projectedEventSerializer.configure(serdeProps, false); - - for (int i = 0; i < numRecords; i++) { - event.eventType = eventTypes[rand.nextInt(eventTypes.length - 1)]; - event.adID = ads.get(rand.nextInt(ads.size() - 1)); - event.eventTime = System.currentTimeMillis(); - final byte[] value = projectedEventSerializer.serialize(topic, event); - producer.send(new ProducerRecord<>(topic, event.adID, value)); - parent.processedRecords++; - parent.processedBytes += value.length + event.adID.length(); - } - } - - final long endTime = System.currentTimeMillis(); - - parent.printResults("Producer Performance [records/latency/rec-sec/MB-sec write]: ", endTime - startTime); - } - - - public void run() { - final int numCampaigns = 100; - final int adsPerCampaign = 10; - - final List ads = new ArrayList<>(numCampaigns * adsPerCampaign); - maybeSetupPhaseCampaigns(campaignsTopic, "simple-benchmark-produce-campaigns", false, numCampaigns, adsPerCampaign, ads); - maybeSetupPhaseEvents(eventsTopic, "simple-benchmark-produce-events", parent.numRecords, ads); - - final CountDownLatch latch = new CountDownLatch(1); - parent.setStreamProperties("simple-benchmark-yahoo" + new Random().nextInt()); - - final KafkaStreams streams = createYahooBenchmarkStreams(parent.props, campaignsTopic, eventsTopic, latch, parent.numRecords); - parent.runGenericBenchmark(streams, "Streams Yahoo Performance [records/latency/rec-sec/MB-sec counted]: ", latch); - - } - // Note: these are also in the streams example package, eventually use 1 file - private class JsonPOJOSerializer implements Serializer { - private final ObjectMapper objectMapper = new ObjectMapper(); - - /** - * Default constructor needed by Kafka - */ - @SuppressWarnings("WeakerAccess") - public JsonPOJOSerializer() {} - - @Override - public byte[] serialize(final String topic, final T data) { - if (data == null) { - return null; - } - - try { - return objectMapper.writeValueAsBytes(data); - } catch (final Exception e) { - throw new SerializationException("Error serializing JSON message", e); - } - } - } - - // Note: these are also in the streams example package, eventuall use 1 file - private class JsonPOJODeserializer implements Deserializer { - private final ObjectMapper objectMapper = new ObjectMapper(); - - private Class tClass; - - /** - * Default constructor needed by Kafka - */ - @SuppressWarnings("WeakerAccess") - public JsonPOJODeserializer() {} - - @SuppressWarnings("unchecked") - @Override - public void configure(final Map props, final boolean isKey) { - tClass = (Class) props.get("JsonPOJOClass"); - } - - @Override - public T deserialize(final String topic, final byte[] bytes) { - if (bytes == null) { - return null; - } - - final T data; - try { - data = objectMapper.readValue(bytes, tClass); - } catch (final Exception e) { - throw new SerializationException(e); - } - - return data; - } - } - - private KafkaStreams createYahooBenchmarkStreams(final Properties streamConfig, final String campaignsTopic, final String eventsTopic, - final CountDownLatch latch, final int numRecords) { - final Map serdeProps = new HashMap<>(); - final Serializer projectedEventSerializer = new JsonPOJOSerializer<>(); - serdeProps.put("JsonPOJOClass", ProjectedEvent.class); - projectedEventSerializer.configure(serdeProps, false); - final Deserializer projectedEventDeserializer = new JsonPOJODeserializer<>(); - serdeProps.put("JsonPOJOClass", ProjectedEvent.class); - projectedEventDeserializer.configure(serdeProps, false); - - final StreamsBuilder builder = new StreamsBuilder(); - final KStream kEvents = builder.stream(eventsTopic, - Consumed.with(Serdes.String(), - Serdes.serdeFrom(projectedEventSerializer, projectedEventDeserializer))); - final KTable kCampaigns = builder.table(campaignsTopic, Consumed.with(Serdes.String(), Serdes.String())); - - final KStream filteredEvents = kEvents - // use peek to quick when last element is processed - .peek((key, value) -> { - parent.processedRecords++; - if (parent.processedRecords % 1000000 == 0) { - System.out.println("Processed " + parent.processedRecords); - } - if (parent.processedRecords >= numRecords) { - latch.countDown(); - } - }) - // only keep "view" events - .filter((key, value) -> value.eventType.equals("view")) - // select just a few of the columns - .mapValues(value -> { - final ProjectedEvent event = new ProjectedEvent(); - event.adID = value.adID; - event.eventTime = value.eventTime; - event.eventType = value.eventType; - return event; - }); - - // deserialize the add ID and campaign ID from the stored value in Kafka - final KTable deserCampaigns = kCampaigns.mapValues(value -> { - final String[] parts = value.split(":"); - final CampaignAd cAdd = new CampaignAd(); - cAdd.adID = parts[0]; - cAdd.campaignID = parts[1]; - return cAdd; - }); - - // join the events with the campaigns - final KStream joined = filteredEvents.join( - deserCampaigns, - (value1, value2) -> value2.campaignID, - Joined.with(Serdes.String(), Serdes.serdeFrom(projectedEventSerializer, projectedEventDeserializer), null) - ); - - // key by campaign rather than by ad as original - final KStream keyedByCampaign = joined - .selectKey((key, value) -> value); - - // calculate windowed counts - keyedByCampaign - .groupByKey(Grouped.with(Serdes.String(), Serdes.String())) - .windowedBy(TimeWindows.of(Duration.ofMillis(10 * 1000))) - .count(Materialized.as("time-windows")); - - return new KafkaStreams(builder.build(), streamConfig); - } -} diff --git a/tests/kafkatest/benchmarks/streams/__init__.py b/tests/kafkatest/benchmarks/streams/__init__.py deleted file mode 100644 index ec2014340d78f..0000000000000 --- a/tests/kafkatest/benchmarks/streams/__init__.py +++ /dev/null @@ -1,14 +0,0 @@ -# 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. diff --git a/tests/kafkatest/benchmarks/streams/streams_simple_benchmark_test.py b/tests/kafkatest/benchmarks/streams/streams_simple_benchmark_test.py deleted file mode 100644 index 2f87f4a06b558..0000000000000 --- a/tests/kafkatest/benchmarks/streams/streams_simple_benchmark_test.py +++ /dev/null @@ -1,164 +0,0 @@ -# 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. - -from ducktape.tests.test import Test -from ducktape.mark.resource import cluster -from ducktape.mark import parametrize, matrix -from kafkatest.tests.kafka_test import KafkaTest - -from kafkatest.services.performance.streams_performance import StreamsSimpleBenchmarkService -from kafkatest.services.zookeeper import ZookeeperService -from kafkatest.services.kafka import KafkaService -from kafkatest.version import DEV_BRANCH - -STREAMS_SIMPLE_TESTS = ["streamprocess", "streamprocesswithsink", "streamprocesswithstatestore", "streamprocesswithwindowstore"] -STREAMS_COUNT_TESTS = ["streamcount", "streamcountwindowed"] -STREAMS_JOIN_TESTS = ["streamtablejoin", "streamstreamjoin", "tabletablejoin"] -NON_STREAMS_TESTS = ["consume", "consumeproduce"] - -ALL_TEST = "all" -STREAMS_SIMPLE_TEST = "streams-simple" -STREAMS_COUNT_TEST = "streams-count" -STREAMS_JOIN_TEST = "streams-join" - - -class StreamsSimpleBenchmarkTest(Test): - """ - Simple benchmark of Kafka Streams. - """ - - def __init__(self, test_context): - super(StreamsSimpleBenchmarkTest, self).__init__(test_context) - - # these values could be updated in ad-hoc benchmarks - self.key_skew = 0 - self.value_size = 1024 - self.num_records = 10000000L - self.num_threads = 1 - - self.replication = 1 - - @cluster(num_nodes=12) - @matrix(test=["consume", "consumeproduce", - "streamprocess", "streamprocesswithsink", "streamprocesswithstatestore", "streamprocesswithwindowstore", - "streamcount", "streamcountwindowed", - "streamtablejoin", "streamstreamjoin", "tabletablejoin"], - scale=[1]) - def test_simple_benchmark(self, test, scale): - """ - Run simple Kafka Streams benchmark - """ - self.driver = [None] * (scale + 1) - - self.final = {} - - ############# - # SETUP PHASE - ############# - self.zk = ZookeeperService(self.test_context, num_nodes=1) - self.zk.start() - self.kafka = KafkaService(self.test_context, num_nodes=scale, zk=self.zk, version=DEV_BRANCH, topics={ - 'simpleBenchmarkSourceTopic1' : { 'partitions': scale, 'replication-factor': self.replication }, - 'simpleBenchmarkSourceTopic2' : { 'partitions': scale, 'replication-factor': self.replication }, - 'simpleBenchmarkSinkTopic' : { 'partitions': scale, 'replication-factor': self.replication }, - 'yahooCampaigns' : { 'partitions': 20, 'replication-factor': self.replication }, - 'yahooEvents' : { 'partitions': 20, 'replication-factor': self.replication } - }) - self.kafka.log_level = "INFO" - self.kafka.start() - - - load_test = "" - if test == ALL_TEST: - load_test = "load-two" - if test in STREAMS_JOIN_TESTS or test == STREAMS_JOIN_TEST: - load_test = "load-two" - if test in STREAMS_COUNT_TESTS or test == STREAMS_COUNT_TEST: - load_test = "load-one" - if test in STREAMS_SIMPLE_TESTS or test == STREAMS_SIMPLE_TEST: - load_test = "load-one" - if test in NON_STREAMS_TESTS: - load_test = "load-one" - - - - ################ - # LOAD PHASE - ################ - self.load_driver = StreamsSimpleBenchmarkService(self.test_context, - self.kafka, - load_test, - self.num_threads, - self.num_records, - self.key_skew, - self.value_size) - - self.load_driver.start() - self.load_driver.wait(3600) # wait at most 30 minutes - self.load_driver.stop() - - if test == ALL_TEST: - for single_test in STREAMS_SIMPLE_TESTS + STREAMS_COUNT_TESTS + STREAMS_JOIN_TESTS: - self.execute(single_test, scale) - elif test == STREAMS_SIMPLE_TEST: - for single_test in STREAMS_SIMPLE_TESTS: - self.execute(single_test, scale) - elif test == STREAMS_COUNT_TEST: - for single_test in STREAMS_COUNT_TESTS: - self.execute(single_test, scale) - elif test == STREAMS_JOIN_TEST: - for single_test in STREAMS_JOIN_TESTS: - self.execute(single_test, scale) - else: - self.execute(test, scale) - - return self.final - - def execute(self, test, scale): - - ################ - # RUN PHASE - ################ - for num in range(0, scale): - self.driver[num] = StreamsSimpleBenchmarkService(self.test_context, - self.kafka, - test, - self.num_threads, - self.num_records, - self.key_skew, - self.value_size) - self.driver[num].start() - - ####################### - # STOP + COLLECT PHASE - ####################### - data = [None] * (scale) - - for num in range(0, scale): - self.driver[num].wait() - self.driver[num].stop() - self.driver[num].node.account.ssh("grep Performance %s" % self.driver[num].STDOUT_FILE, allow_fail=False) - data[num] = self.driver[num].collect_data(self.driver[num].node, "") - self.driver[num].read_jmx_output_all_nodes() - - for num in range(0, scale): - for key in data[num]: - self.final[key + "-" + str(num)] = data[num][key] - - for key in sorted(self.driver[num].jmx_stats[0]): - self.logger.info("%s: %s" % (key, self.driver[num].jmx_stats[0][key])) - - self.final[test + "-jmx-avg-" + str(num)] = self.driver[num].average_jmx_value - self.final[test + "-jmx-max-" + str(num)] = self.driver[num].maximum_jmx_value diff --git a/tests/kafkatest/services/performance/streams_performance.py b/tests/kafkatest/services/performance/streams_performance.py deleted file mode 100644 index 049c272afd941..0000000000000 --- a/tests/kafkatest/services/performance/streams_performance.py +++ /dev/null @@ -1,108 +0,0 @@ -# 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. - -from kafkatest.services.monitor.jmx import JmxMixin -from kafkatest.services.streams import StreamsTestBaseService -from kafkatest.services.kafka import KafkaConfig -from kafkatest.services import streams_property - -# -# Class used to start the simple Kafka Streams benchmark -# - -class StreamsSimpleBenchmarkService(StreamsTestBaseService): - """Base class for simple Kafka Streams benchmark""" - - def __init__(self, test_context, kafka, test_name, num_threads, num_recs_or_wait_ms, key_skew, value_size): - super(StreamsSimpleBenchmarkService, self).__init__(test_context, - kafka, - "org.apache.kafka.streams.perf.SimpleBenchmark", - test_name, - num_recs_or_wait_ms, - key_skew, - value_size) - - self.jmx_option = "" - if test_name.startswith('stream') or test_name.startswith('table'): - self.jmx_option = "stream-jmx" - JmxMixin.__init__(self, - num_nodes=1, - jmx_object_names=['kafka.streams:type=stream-thread-metrics,thread-id=simple-benchmark-StreamThread-%d' %(i+1) for i in range(num_threads)], - jmx_attributes=['process-latency-avg', - 'process-rate', - 'commit-latency-avg', - 'commit-rate', - 'poll-latency-avg', - 'poll-rate'], - root=StreamsTestBaseService.PERSISTENT_ROOT) - - if test_name.startswith('consume'): - self.jmx_option = "consumer-jmx" - JmxMixin.__init__(self, - num_nodes=1, - jmx_object_names=['kafka.consumer:type=consumer-fetch-manager-metrics,client-id=simple-benchmark-consumer'], - jmx_attributes=['records-consumed-rate'], - root=StreamsTestBaseService.PERSISTENT_ROOT) - - self.num_threads = num_threads - - def prop_file(self): - cfg = KafkaConfig(**{streams_property.STATE_DIR: self.PERSISTENT_ROOT, - streams_property.KAFKA_SERVERS: self.kafka.bootstrap_servers(), - streams_property.NUM_THREADS: self.num_threads}) - return cfg.render() - - - def start_cmd(self, node): - if self.jmx_option != "": - args = self.args.copy() - args['jmx_port'] = self.jmx_port - args['config_file'] = self.CONFIG_FILE - args['stdout'] = self.STDOUT_FILE - args['stderr'] = self.STDERR_FILE - args['pidfile'] = self.PID_FILE - args['log4j'] = self.LOG4J_CONFIG_FILE - args['kafka_run_class'] = self.path.script("kafka-run-class.sh", node) - - cmd = "( export JMX_PORT=%(jmx_port)s; export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%(log4j)s\"; " \ - "INCLUDE_TEST_JARS=true %(kafka_run_class)s %(streams_class_name)s " \ - " %(config_file)s %(user_test_args1)s %(user_test_args2)s %(user_test_args3)s" \ - " %(user_test_args4)s & echo $! >&3 ) 1>> %(stdout)s 2>> %(stderr)s 3> %(pidfile)s" % args - - else: - cmd = super(StreamsSimpleBenchmarkService, self).start_cmd(node) - - return cmd - - def start_node(self, node): - super(StreamsSimpleBenchmarkService, self).start_node(node) - - if self.jmx_option != "": - self.start_jmx_tool(1, node) - - def clean_node(self, node): - if self.jmx_option != "": - JmxMixin.clean_node(self, node) - - super(StreamsSimpleBenchmarkService, self).clean_node(node) - - def collect_data(self, node, tag = None): - # Collect the data and return it to the framework - output = node.account.ssh_capture("grep Performance %s" % self.STDOUT_FILE) - data = {} - for line in output: - parts = line.split(':') - data[tag + parts[0]] = parts[1] - return data From b926ccbbd1395497354d9836e55cc84a5a8ce443 Mon Sep 17 00:00:00 2001 From: Alex Diachenko Date: Mon, 5 Oct 2020 17:24:44 -0700 Subject: [PATCH 09/15] KAFKA-10439: Connect's Values to parse BigInteger as Decimal with zero scale. (#9320) The `org.apache.kafka.connect.data.Values#parse` method parses integers, which are larger than `Long.MAX_VALUE` as `double` with `Schema.FLOAT64_SCHEMA`. That means we are losing precision for these larger integers. For example: `SchemaAndValue schemaAndValue = Values.parseString("9223372036854775808");` returns: `SchemaAndValue{schema=Schema{FLOAT64}, value=9.223372036854776E18}` Also, this method parses values that can be parsed as `FLOAT32` to `FLOAT64`. This PR changes parsing logic, to use `FLOAT32`/`FLOAT64` for numbers that don't have fraction part(`decimal.scale()!=0`) only, and use an arbitrary-precision `org.apache.kafka.connect.data.Decimal` otherwise. Also, it updates the method to parse numbers, that can be represented as `float` to `FLOAT64`. Added unit tests, that cover parsing `BigInteger`, `Byte`, `Short`, `Integer`, `Long`, `Float`, `Double` types. Reviewers: Konstantine Karantasis --- .../org/apache/kafka/connect/data/Values.java | 8 +- .../apache/kafka/connect/data/ValuesTest.java | 128 ++++++++++++++++++ 2 files changed, 135 insertions(+), 1 deletion(-) diff --git a/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java b/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java index d99fbcabf86df..2c0d6bae4fd0b 100644 --- a/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java +++ b/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java @@ -943,8 +943,14 @@ protected static SchemaAndValue parse(Parser parser, boolean embedded) throws No } catch (ArithmeticException e) { // continue } + float fValue = decimal.floatValue(); + if (fValue != Float.NEGATIVE_INFINITY && fValue != Float.POSITIVE_INFINITY + && decimal.scale() != 0) { + return new SchemaAndValue(Schema.FLOAT32_SCHEMA, fValue); + } double dValue = decimal.doubleValue(); - if (dValue != Double.NEGATIVE_INFINITY && dValue != Double.POSITIVE_INFINITY) { + if (dValue != Double.NEGATIVE_INFINITY && dValue != Double.POSITIVE_INFINITY + && decimal.scale() != 0) { return new SchemaAndValue(Schema.FLOAT64_SCHEMA, dValue); } Schema schema = Decimal.schema(decimal.scale()); diff --git a/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java b/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java index c437e46c25956..63c05eaa9b1e3 100644 --- a/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java +++ b/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java @@ -21,6 +21,8 @@ import org.apache.kafka.connect.errors.DataException; import org.junit.Test; +import java.math.BigDecimal; +import java.math.BigInteger; import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Arrays; @@ -726,6 +728,132 @@ public void shouldConvertTimestampValues() { public void canConsume() { } + @Test + public void shouldParseBigIntegerAsDecimalWithZeroScale() { + BigInteger value = BigInteger.valueOf(Long.MAX_VALUE).add(new BigInteger("1")); + SchemaAndValue schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Decimal.schema(0), schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof BigDecimal); + assertEquals(value, ((BigDecimal) schemaAndValue.value()).unscaledValue()); + value = BigInteger.valueOf(Long.MIN_VALUE).subtract(new BigInteger("1")); + schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Decimal.schema(0), schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof BigDecimal); + assertEquals(value, ((BigDecimal) schemaAndValue.value()).unscaledValue()); + } + + @Test + public void shouldParseByteAsInt8() { + Byte value = Byte.MAX_VALUE; + SchemaAndValue schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.INT8_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Byte); + assertEquals(value.byteValue(), ((Byte) schemaAndValue.value()).byteValue()); + value = Byte.MIN_VALUE; + schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.INT8_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Byte); + assertEquals(value.byteValue(), ((Byte) schemaAndValue.value()).byteValue()); + } + + @Test + public void shouldParseShortAsInt16() { + Short value = Short.MAX_VALUE; + SchemaAndValue schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.INT16_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Short); + assertEquals(value.shortValue(), ((Short) schemaAndValue.value()).shortValue()); + value = Short.MIN_VALUE; + schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.INT16_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Short); + assertEquals(value.shortValue(), ((Short) schemaAndValue.value()).shortValue()); + } + + @Test + public void shouldParseIntegerAsInt32() { + Integer value = Integer.MAX_VALUE; + SchemaAndValue schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.INT32_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Integer); + assertEquals(value.intValue(), ((Integer) schemaAndValue.value()).intValue()); + value = Integer.MIN_VALUE; + schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.INT32_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Integer); + assertEquals(value.intValue(), ((Integer) schemaAndValue.value()).intValue()); + } + + @Test + public void shouldParseLongAsInt64() { + Long value = Long.MAX_VALUE; + SchemaAndValue schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.INT64_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Long); + assertEquals(value.longValue(), ((Long) schemaAndValue.value()).longValue()); + value = Long.MIN_VALUE; + schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.INT64_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Long); + assertEquals(value.longValue(), ((Long) schemaAndValue.value()).longValue()); + } + + @Test + public void shouldParseFloatAsFloat32() { + Float value = Float.MAX_VALUE; + SchemaAndValue schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.FLOAT32_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Float); + assertEquals(value.floatValue(), ((Float) schemaAndValue.value()).floatValue(), 0); + value = -Float.MAX_VALUE; + schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.FLOAT32_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Float); + assertEquals(value.floatValue(), ((Float) schemaAndValue.value()).floatValue(), 0); + } + + @Test + public void shouldParseDoubleAsFloat64() { + Double value = Double.MAX_VALUE; + SchemaAndValue schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.FLOAT64_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Double); + assertEquals(value.doubleValue(), ((Double) schemaAndValue.value()).doubleValue(), 0); + value = -Double.MAX_VALUE; + schemaAndValue = Values.parseString( + String.valueOf(value) + ); + assertEquals(Schema.FLOAT64_SCHEMA, schemaAndValue.schema()); + assertTrue(schemaAndValue.value() instanceof Double); + assertEquals(value.doubleValue(), ((Double) schemaAndValue.value()).doubleValue(), 0); + } + protected void assertParsed(String input) { assertParsed(input, input); } From 47c517931cc7433bfad0eb1c4d2f800766c02fae Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Tue, 6 Oct 2020 14:18:54 -0400 Subject: [PATCH 10/15] KAFKA-10188: Prevent SinkTask::preCommit from being called after SinkTask::stop (#8910) --- .../kafka/connect/runtime/WorkerSinkTask.java | 15 +++--- .../connect/runtime/WorkerSinkTaskTest.java | 50 +++++++++++++++++++ 2 files changed, 58 insertions(+), 7 deletions(-) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java index 5db4030d9483c..54b19b6a485c8 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java @@ -93,6 +93,7 @@ class WorkerSinkTask extends WorkerTask { private int commitFailures; private boolean pausedForRedelivery; private boolean committing; + private boolean taskStopped; public WorkerSinkTask(ConnectorTaskId id, SinkTask task, @@ -135,6 +136,7 @@ public WorkerSinkTask(ConnectorTaskId id, this.sinkTaskMetricsGroup.recordOffsetSequenceNumber(commitSeqno); this.consumer = consumer; this.isTopicTrackingEnabled = workerConfig.getBoolean(TOPIC_TRACKING_ENABLE_CONFIG); + this.taskStopped = false; } @Override @@ -164,13 +166,8 @@ protected void close() { } catch (Throwable t) { log.warn("Could not stop task", t); } - if (consumer != null) { - try { - consumer.close(); - } catch (Throwable t) { - log.warn("Could not close consumer", t); - } - } + taskStopped = true; + Utils.closeQuietly(consumer, "consumer"); try { transformationChain.close(); } catch (Throwable t) { @@ -672,6 +669,10 @@ else if (!context.pausedPartitions().isEmpty()) @Override public void onPartitionsRevoked(Collection partitions) { + if (taskStopped) { + log.trace("Skipping partition revocation callback as task has already been stopped"); + return; + } log.debug("{} Partitions revoked", WorkerSinkTask.this); try { closePartitions(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java index 5dc2f44fecbd3..98cfa7e80ba6b 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java @@ -315,6 +315,56 @@ public void testPause() throws Exception { PowerMock.verifyAll(); } + @Test + public void testShutdown() throws Exception { + createTask(initialState); + + expectInitializeTask(); + expectTaskGetTopic(true); + + // first iteration + expectPollInitialAssignment(); + + // second iteration + EasyMock.expect(sinkTask.preCommit(EasyMock.anyObject())).andReturn(Collections.emptyMap()); + expectConsumerPoll(1); + expectConversionAndTransformation(1); + sinkTask.put(EasyMock.>anyObject()); + EasyMock.expectLastCall(); + + // WorkerSinkTask::stop + consumer.wakeup(); + PowerMock.expectLastCall(); + sinkTask.stop(); + PowerMock.expectLastCall(); + + // WorkerSinkTask::close + consumer.close(); + PowerMock.expectLastCall().andAnswer(new IAnswer() { + @Override + public Object answer() throws Throwable { + rebalanceListener.getValue().onPartitionsRevoked( + asList(TOPIC_PARTITION, TOPIC_PARTITION2) + ); + return null; + } + }); + transformationChain.close(); + PowerMock.expectLastCall(); + + PowerMock.replayAll(); + + workerTask.initialize(TASK_CONFIG); + workerTask.initializeAndStart(); + workerTask.iteration(); + sinkTaskContext.getValue().requestCommit(); // Force an offset commit + workerTask.iteration(); + workerTask.stop(); + workerTask.close(); + + PowerMock.verifyAll(); + } + @Test public void testPollRedelivery() throws Exception { createTask(initialState); From 673eaea10e298a0bf4822c9e38bab1d816799dd6 Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Mon, 12 Oct 2020 12:42:42 -0400 Subject: [PATCH 11/15] KAFKA-10574: Fix infinite loop in Values::parseString (#9375) Fix infinite loop in Values::parseString Author: Chris Egerton Reviewers: Konstantine Karantasis , Randall Hauch --- .../java/org/apache/kafka/connect/data/Values.java | 2 +- .../org/apache/kafka/connect/data/ValuesTest.java | 13 +++++++++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java b/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java index 2c0d6bae4fd0b..2e30a7e9799b2 100644 --- a/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java +++ b/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java @@ -1193,7 +1193,7 @@ private String consumeNextToken() throws NoSuchElementException { boolean escaped = false; int start = iter.getIndex(); char c = iter.current(); - while (c != CharacterIterator.DONE) { + while (canConsumeNextToken()) { switch (c) { case '\\': escaped = !escaped; diff --git a/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java b/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java index 63c05eaa9b1e3..cb6f71a60dfa3 100644 --- a/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java +++ b/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java @@ -23,6 +23,7 @@ import java.math.BigDecimal; import java.math.BigInteger; +import java.nio.charset.StandardCharsets; import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Arrays; @@ -75,6 +76,18 @@ public class ValuesTest { INT_LIST.add(-987654321); } + @Test(timeout = 5000) + public void shouldNotEncounterInfiniteLoop() { + // This byte sequence gets parsed as CharacterIterator.DONE and can cause issues if + // comparisons to that character are done to check if the end of a string has been reached. + // For more information, see https://issues.apache.org/jira/browse/KAFKA-10574 + byte[] bytes = new byte[] {-17, -65, -65}; + String str = new String(bytes, StandardCharsets.UTF_8); + SchemaAndValue schemaAndValue = Values.parseString(str); + assertEquals(Type.STRING, schemaAndValue.schema().type()); + assertEquals(str, schemaAndValue.value()); + } + @Test public void shouldNotParseUnquotedEmbeddedMapKeysAsStrings() { SchemaAndValue schemaAndValue = Values.parseString("{foo: 3}"); From 6688ea2235b1fcf18b22e270a3c63e4abf0daea7 Mon Sep 17 00:00:00 2001 From: Randall Hauch Date: Fri, 16 Oct 2020 09:14:43 -0500 Subject: [PATCH 12/15] =?UTF-8?q?KAFKA-10600:=20Connect=20should=20not=20a?= =?UTF-8?q?dd=20error=20to=20connector=20validation=20values=20for=20prope?= =?UTF-8?q?rties=20not=20in=20connector=E2=80=99s=20ConfigDef=20(#9425)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Connect should not always add an error to configuration values in validation results that don't have a `ConfigKey` defined in the connector's `ConfigDef`, and any errors on such configuration values included by the connector should be counted in the total number of errors. Added more unit tests for `AbstractHerder.generateResult(...)`. Author: Randall Hauch Reviewer: Konstantine Karantasis --- .../kafka/connect/runtime/AbstractHerder.java | 2 +- .../connect/runtime/AbstractHerderTest.java | 173 +++++++++++++++++- 2 files changed, 173 insertions(+), 2 deletions(-) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java index 04142d4c43242..50c98f7ac739b 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java @@ -475,8 +475,8 @@ public static ConfigInfos generateResult(String connType, Map String configName = configValue.name(); configValueMap.put(configName, configValue); if (!configKeys.containsKey(configName)) { - configValue.addErrorMessage("Configuration is not defined: " + configName); configInfoList.add(new ConfigInfo(null, convertConfigValue(configValue, null))); + errorCount += configValue.errorMessages().size(); } } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java index fe57d57972c39..6a96a51fcb871 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java @@ -21,6 +21,7 @@ import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.config.ConfigTransformer; +import org.apache.kafka.common.config.ConfigValue; import org.apache.kafka.common.config.SaslConfigs; import org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler; import org.apache.kafka.connect.connector.ConnectRecord; @@ -34,6 +35,7 @@ import org.apache.kafka.connect.runtime.isolation.Plugins; import org.apache.kafka.connect.runtime.rest.entities.ConfigInfo; import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos; +import org.apache.kafka.connect.runtime.rest.entities.ConfigValueInfo; import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo; import org.apache.kafka.connect.runtime.rest.entities.ConnectorType; import org.apache.kafka.connect.runtime.rest.errors.BadRequestException; @@ -64,6 +66,7 @@ import java.util.stream.Collectors; import static org.apache.kafka.connect.runtime.AbstractHerder.keysWithVariableValues; +import static org.junit.Assert.assertNull; import static org.powermock.api.easymock.PowerMock.verifyAll; import static org.powermock.api.easymock.PowerMock.replayAll; import static org.easymock.EasyMock.strictMock; @@ -522,10 +525,178 @@ public void testConfigProviderRegex() { testConfigProviderRegex("plain.PlainLoginModule required username", false); } + @Test + public void testGenerateResultWithConfigValuesAllUsingConfigKeysAndWithNoErrors() { + String name = "com.acme.connector.MyConnector"; + Map keys = new HashMap<>(); + addConfigKey(keys, "config.a1", null); + addConfigKey(keys, "config.b1", "group B"); + addConfigKey(keys, "config.b2", "group B"); + addConfigKey(keys, "config.c1", "group C"); + + List groups = Arrays.asList("groupB", "group C"); + List values = new ArrayList<>(); + addValue(values, "config.a1", "value.a1"); + addValue(values, "config.b1", "value.b1"); + addValue(values, "config.b2", "value.b2"); + addValue(values, "config.c1", "value.c1"); + + ConfigInfos infos = AbstractHerder.generateResult(name, keys, values, groups); + assertEquals(name, infos.name()); + assertEquals(groups, infos.groups()); + assertEquals(values.size(), infos.values().size()); + assertEquals(0, infos.errorCount()); + assertInfoKey(infos, "config.a1", null); + assertInfoKey(infos, "config.b1", "group B"); + assertInfoKey(infos, "config.b2", "group B"); + assertInfoKey(infos, "config.c1", "group C"); + assertInfoValue(infos, "config.a1", "value.a1"); + assertInfoValue(infos, "config.b1", "value.b1"); + assertInfoValue(infos, "config.b2", "value.b2"); + assertInfoValue(infos, "config.c1", "value.c1"); + } + + @Test + public void testGenerateResultWithConfigValuesAllUsingConfigKeysAndWithSomeErrors() { + String name = "com.acme.connector.MyConnector"; + Map keys = new HashMap<>(); + addConfigKey(keys, "config.a1", null); + addConfigKey(keys, "config.b1", "group B"); + addConfigKey(keys, "config.b2", "group B"); + addConfigKey(keys, "config.c1", "group C"); + + List groups = Arrays.asList("groupB", "group C"); + List values = new ArrayList<>(); + addValue(values, "config.a1", "value.a1"); + addValue(values, "config.b1", "value.b1"); + addValue(values, "config.b2", "value.b2"); + addValue(values, "config.c1", "value.c1", "error c1"); + + ConfigInfos infos = AbstractHerder.generateResult(name, keys, values, groups); + assertEquals(name, infos.name()); + assertEquals(groups, infos.groups()); + assertEquals(values.size(), infos.values().size()); + assertEquals(1, infos.errorCount()); + assertInfoKey(infos, "config.a1", null); + assertInfoKey(infos, "config.b1", "group B"); + assertInfoKey(infos, "config.b2", "group B"); + assertInfoKey(infos, "config.c1", "group C"); + assertInfoValue(infos, "config.a1", "value.a1"); + assertInfoValue(infos, "config.b1", "value.b1"); + assertInfoValue(infos, "config.b2", "value.b2"); + assertInfoValue(infos, "config.c1", "value.c1", "error c1"); + } + + @Test + public void testGenerateResultWithConfigValuesMoreThanConfigKeysAndWithSomeErrors() { + String name = "com.acme.connector.MyConnector"; + Map keys = new HashMap<>(); + addConfigKey(keys, "config.a1", null); + addConfigKey(keys, "config.b1", "group B"); + addConfigKey(keys, "config.b2", "group B"); + addConfigKey(keys, "config.c1", "group C"); + + List groups = Arrays.asList("groupB", "group C"); + List values = new ArrayList<>(); + addValue(values, "config.a1", "value.a1"); + addValue(values, "config.b1", "value.b1"); + addValue(values, "config.b2", "value.b2"); + addValue(values, "config.c1", "value.c1", "error c1"); + addValue(values, "config.extra1", "value.extra1"); + addValue(values, "config.extra2", "value.extra2", "error extra2"); + + ConfigInfos infos = AbstractHerder.generateResult(name, keys, values, groups); + assertEquals(name, infos.name()); + assertEquals(groups, infos.groups()); + assertEquals(values.size(), infos.values().size()); + assertEquals(2, infos.errorCount()); + assertInfoKey(infos, "config.a1", null); + assertInfoKey(infos, "config.b1", "group B"); + assertInfoKey(infos, "config.b2", "group B"); + assertInfoKey(infos, "config.c1", "group C"); + assertNoInfoKey(infos, "config.extra1"); + assertNoInfoKey(infos, "config.extra2"); + assertInfoValue(infos, "config.a1", "value.a1"); + assertInfoValue(infos, "config.b1", "value.b1"); + assertInfoValue(infos, "config.b2", "value.b2"); + assertInfoValue(infos, "config.c1", "value.c1", "error c1"); + assertInfoValue(infos, "config.extra1", "value.extra1"); + assertInfoValue(infos, "config.extra2", "value.extra2", "error extra2"); + } + + @Test + public void testGenerateResultWithConfigValuesWithNoConfigKeysAndWithSomeErrors() { + String name = "com.acme.connector.MyConnector"; + Map keys = new HashMap<>(); + + List groups = new ArrayList<>(); + List values = new ArrayList<>(); + addValue(values, "config.a1", "value.a1"); + addValue(values, "config.b1", "value.b1"); + addValue(values, "config.b2", "value.b2"); + addValue(values, "config.c1", "value.c1", "error c1"); + addValue(values, "config.extra1", "value.extra1"); + addValue(values, "config.extra2", "value.extra2", "error extra2"); + + ConfigInfos infos = AbstractHerder.generateResult(name, keys, values, groups); + assertEquals(name, infos.name()); + assertEquals(groups, infos.groups()); + assertEquals(values.size(), infos.values().size()); + assertEquals(2, infos.errorCount()); + assertNoInfoKey(infos, "config.a1"); + assertNoInfoKey(infos, "config.b1"); + assertNoInfoKey(infos, "config.b2"); + assertNoInfoKey(infos, "config.c1"); + assertNoInfoKey(infos, "config.extra1"); + assertNoInfoKey(infos, "config.extra2"); + assertInfoValue(infos, "config.a1", "value.a1"); + assertInfoValue(infos, "config.b1", "value.b1"); + assertInfoValue(infos, "config.b2", "value.b2"); + assertInfoValue(infos, "config.c1", "value.c1", "error c1"); + assertInfoValue(infos, "config.extra1", "value.extra1"); + assertInfoValue(infos, "config.extra2", "value.extra2", "error extra2"); + } + + protected void addConfigKey(Map keys, String name, String group) { + keys.put(name, new ConfigDef.ConfigKey(name, ConfigDef.Type.STRING, null, null, + ConfigDef.Importance.HIGH, "doc", group, 10, + ConfigDef.Width.MEDIUM, "display name", Collections.emptyList(), null, false)); + } + + protected void addValue(List values, String name, String value, String...errors) { + values.add(new ConfigValue(name, value, new ArrayList<>(), Arrays.asList(errors))); + } + + protected void assertInfoKey(ConfigInfos infos, String name, String group) { + ConfigInfo info = findInfo(infos, name); + assertEquals(name, info.configKey().name()); + assertEquals(group, info.configKey().group()); + } + + protected void assertNoInfoKey(ConfigInfos infos, String name) { + ConfigInfo info = findInfo(infos, name); + assertNull(info.configKey()); + } + + protected void assertInfoValue(ConfigInfos infos, String name, String value, String...errors) { + ConfigValueInfo info = findInfo(infos, name).configValue(); + assertEquals(name, info.name()); + assertEquals(value, info.value()); + assertEquals(Arrays.asList(errors), info.errors()); + } + + protected ConfigInfo findInfo(ConfigInfos infos, String name) { + return infos.values() + .stream() + .filter(i -> i.configValue().name().equals(name)) + .findFirst() + .orElse(null); + } + private void testConfigProviderRegex(String rawConnConfig) { testConfigProviderRegex(rawConnConfig, true); } - + private void testConfigProviderRegex(String rawConnConfig, boolean expected) { Set keys = keysWithVariableValues(Collections.singletonMap("key", rawConnConfig), ConfigTransformer.DEFAULT_PATTERN); boolean actual = keys != null && !keys.isEmpty() && keys.contains("key"); From 71f7ed3923c802cb949b80e28e6a255ac0fa046d Mon Sep 17 00:00:00 2001 From: Mickael Maison Date: Mon, 19 Oct 2020 17:51:44 +0200 Subject: [PATCH 13/15] KAFKA-10332: Update MM2 refreshTopicPartitions() logic (#9343) Trigger task reconfiguration when: - topic-partitions are created or deleted on source cluster - topic-partitions are missing on target cluster Authors: Mickael Maison , Edoardo Comar Reviewer: Randall Hauch --- .../connect/mirror/MirrorSourceConnector.java | 54 +++++++++++++------ .../mirror/MirrorSourceConnectorTest.java | 31 ++++++++++- 2 files changed, 67 insertions(+), 18 deletions(-) diff --git a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConnector.java b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConnector.java index 041814affb2aa..0f6eb46dd5b4d 100644 --- a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConnector.java +++ b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConnector.java @@ -96,7 +96,7 @@ public MirrorSourceConnector() { this.replicationPolicy = replicationPolicy; this.topicFilter = topicFilter; this.configPropertyFilter = configPropertyFilter; - } + } @Override public void start(Map props) { @@ -202,6 +202,7 @@ List findTargetTopicPartitions() throws InterruptedException, ExecutionException { Set topics = listTopics(targetAdminClient).stream() .filter(t -> sourceAndTarget.source().equals(replicationPolicy.topicSource(t))) + .filter(t -> !t.equals(config.checkpointsTopic())) .collect(Collectors.toSet()); return describeTopics(targetAdminClient, topics).stream() .flatMap(MirrorSourceConnector::expandTopicDescription) @@ -211,23 +212,44 @@ List findTargetTopicPartitions() // visible for testing void refreshTopicPartitions() throws InterruptedException, ExecutionException { - knownSourceTopicPartitions = findSourceTopicPartitions(); - knownTargetTopicPartitions = findTargetTopicPartitions(); - List upstreamTargetTopicPartitions = knownTargetTopicPartitions.stream() + + List sourceTopicPartitions = findSourceTopicPartitions(); + List targetTopicPartitions = findTargetTopicPartitions(); + + Set sourceTopicPartitionsSet = new HashSet<>(sourceTopicPartitions); + Set knownSourceTopicPartitionsSet = new HashSet<>(knownSourceTopicPartitions); + + Set upstreamTargetTopicPartitions = targetTopicPartitions.stream() .map(x -> new TopicPartition(replicationPolicy.upstreamTopic(x.topic()), x.partition())) - .collect(Collectors.toList()); + .collect(Collectors.toSet()); + + Set missingInTarget = new HashSet<>(sourceTopicPartitions); + missingInTarget.removeAll(upstreamTargetTopicPartitions); + + knownTargetTopicPartitions = targetTopicPartitions; + + // Detect if topic-partitions were added or deleted from the source cluster + // or if topic-partitions are missing from the target cluster + if (!knownSourceTopicPartitionsSet.equals(sourceTopicPartitionsSet) || !missingInTarget.isEmpty()) { + + Set newTopicPartitions = sourceTopicPartitionsSet; + newTopicPartitions.removeAll(knownSourceTopicPartitions); + + Set deletedTopicPartitions = knownSourceTopicPartitionsSet; + deletedTopicPartitions.removeAll(sourceTopicPartitions); + + log.info("Found {} new topic-partitions on {}. " + + "Found {} deleted topic-partitions on {}. " + + "Found {} topic-partitions missing on {}.", + newTopicPartitions.size(), sourceAndTarget.source(), + deletedTopicPartitions.size(), sourceAndTarget.source(), + missingInTarget.size(), sourceAndTarget.target()); + + log.trace("Found new topic-partitions on {}: {}", sourceAndTarget.source(), newTopicPartitions); + log.trace("Found deleted topic-partitions on {}: {}", sourceAndTarget.source(), deletedTopicPartitions); + log.trace("Found missing topic-partitions on {}: {}", sourceAndTarget.target(), missingInTarget); - Set newTopicPartitions = new HashSet<>(); - newTopicPartitions.addAll(knownSourceTopicPartitions); - newTopicPartitions.removeAll(upstreamTargetTopicPartitions); - Set deadTopicPartitions = new HashSet<>(); - deadTopicPartitions.addAll(upstreamTargetTopicPartitions); - deadTopicPartitions.removeAll(knownSourceTopicPartitions); - if (!newTopicPartitions.isEmpty() || !deadTopicPartitions.isEmpty()) { - log.info("Found {} topic-partitions on {}. {} are new. {} were removed. Previously had {}.", - knownSourceTopicPartitions.size(), sourceAndTarget.source(), newTopicPartitions.size(), - deadTopicPartitions.size(), knownSourceTopicPartitions.size()); - log.trace("Found new topic-partitions: {}", newTopicPartitions); + knownSourceTopicPartitions = sourceTopicPartitions; computeAndCreateTopicPartitions(); context.requestTaskReconfiguration(); } diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConnectorTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConnectorTest.java index e86d21ee67e6c..c915845a606a1 100644 --- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConnectorTest.java +++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConnectorTest.java @@ -186,7 +186,7 @@ public void testRefreshTopicPartitions() throws Exception { connector.initialize(mock(ConnectorContext.class)); connector = spy(connector); - List sourceTopicPartitions = Arrays.asList(new TopicPartition("topic", 0)); + List sourceTopicPartitions = Collections.singletonList(new TopicPartition("topic", 0)); doReturn(sourceTopicPartitions).when(connector).findSourceTopicPartitions(); doReturn(Collections.emptyList()).when(connector).findTargetTopicPartitions(); doNothing().when(connector).createTopicPartitions(any(), any(), any()); @@ -205,11 +205,38 @@ public void testRefreshTopicPartitions() throws Exception { eq(expectedNewTopics), eq(Collections.emptyMap())); - List targetTopicPartitions = Arrays.asList(new TopicPartition("source.topic", 0)); + List targetTopicPartitions = Collections.singletonList(new TopicPartition("source.topic", 0)); doReturn(targetTopicPartitions).when(connector).findTargetTopicPartitions(); connector.refreshTopicPartitions(); // once target topic is created, refreshTopicPartitions() will NOT call computeAndCreateTopicPartitions() again verify(connector, times(2)).computeAndCreateTopicPartitions(); } + + @Test + public void testRefreshTopicPartitionsTopicOnTargetFirst() throws Exception { + MirrorSourceConnector connector = new MirrorSourceConnector(new SourceAndTarget("source", "target"), + new DefaultReplicationPolicy(), new DefaultTopicFilter(), new DefaultConfigPropertyFilter()); + connector.initialize(mock(ConnectorContext.class)); + connector = spy(connector); + + List sourceTopicPartitions = Collections.emptyList(); + List targetTopicPartitions = Collections.singletonList(new TopicPartition("source.topic", 0)); + doReturn(sourceTopicPartitions).when(connector).findSourceTopicPartitions(); + doReturn(targetTopicPartitions).when(connector).findTargetTopicPartitions(); + doNothing().when(connector).createTopicPartitions(any(), any(), any()); + + // partitions appearing on the target cluster should not cause reconfiguration + connector.refreshTopicPartitions(); + connector.refreshTopicPartitions(); + verify(connector, times(0)).computeAndCreateTopicPartitions(); + + sourceTopicPartitions = Collections.singletonList(new TopicPartition("topic", 0)); + doReturn(sourceTopicPartitions).when(connector).findSourceTopicPartitions(); + + // when partitions are added to the source cluster, reconfiguration is triggered + connector.refreshTopicPartitions(); + verify(connector, times(1)).computeAndCreateTopicPartitions(); + + } } From 28c8cc59925fc864e789e8a0a04f9f616baac67a Mon Sep 17 00:00:00 2001 From: Justine Olshan Date: Tue, 20 Oct 2020 07:00:21 -0700 Subject: [PATCH 14/15] MINOR: Fix now that kafka.apache.org resolves to 3 IP addresses (#9305) ClusterConnectStatesTest and ClientUtilsTest were failing because they expected kafka.apache.org to resolve to 2 IP addresses. This updates the tests so they reflect that DNS resolves to 3 addresses. Reviewers: Ismael Juma --- .../java/org/apache/kafka/clients/ClientUtilsTest.java | 2 +- .../kafka/clients/ClusterConnectionStatesTest.java | 9 ++++----- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java b/clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java index 5dd65ed86e3c1..572896f56bd11 100644 --- a/clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java @@ -107,7 +107,7 @@ public void testResolveDnsLookup() throws UnknownHostException { @Test public void testResolveDnsLookupAllIps() throws UnknownHostException { - assertEquals(2, ClientUtils.resolve("kafka.apache.org", ClientDnsLookup.USE_ALL_DNS_IPS).size()); + assertTrue(ClientUtils.resolve("kafka.apache.org", ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1); } private List checkWithoutLookup(String... url) { diff --git a/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java b/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java index 2a427cc5bad48..fbb449715389f 100644 --- a/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java @@ -256,7 +256,7 @@ public void testSingleIPWithUseAll() throws UnknownHostException { @Test public void testMultipleIPsWithDefault() throws UnknownHostException { - assertEquals(2, ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size()); + assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1); connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.DEFAULT); InetAddress currAddress = connectionStates.currentAddress(nodeId1); @@ -266,22 +266,21 @@ public void testMultipleIPsWithDefault() throws UnknownHostException { @Test public void testMultipleIPsWithUseAll() throws UnknownHostException { - assertEquals(2, ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size()); + assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1); connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS); InetAddress addr1 = connectionStates.currentAddress(nodeId1); connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS); InetAddress addr2 = connectionStates.currentAddress(nodeId1); assertNotSame(addr1, addr2); - connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS); InetAddress addr3 = connectionStates.currentAddress(nodeId1); - assertSame(addr1, addr3); + assertNotSame(addr1, addr3); } @Test public void testHostResolveChange() throws UnknownHostException, ReflectiveOperationException { - assertEquals(2, ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size()); + assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1); connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.DEFAULT); InetAddress addr1 = connectionStates.currentAddress(nodeId1); From fbe40539710650c015d762698c6d34f2ba128658 Mon Sep 17 00:00:00 2001 From: David Arthur Date: Thu, 1 Oct 2020 14:03:36 -0400 Subject: [PATCH 15/15] Fix a Scala 2.11 compile error in GroupMetadataManagerTest This was a cherry-pick for the fix in core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala from 63f3e1c33b03870c5d4fe2dba40cf385cfab6dca --- .../coordinator/group/GroupMetadataManagerTest.scala | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala index af34b9e5e1f1d..f8a3faf8c6161 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala @@ -45,7 +45,7 @@ import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse import org.apache.kafka.common.utils.Utils import org.apache.kafka.common.KafkaException import org.easymock.{Capture, EasyMock, IAnswer} -import org.junit.Assert.{assertEquals, assertFalse, assertNull, assertTrue, assertThrows} +import org.junit.Assert.{assertEquals, assertFalse, assertNull, assertTrue} import org.junit.{Before, Test} import org.scalatest.Assertions.fail @@ -922,9 +922,13 @@ class GroupMetadataManagerTest { // reset the position to the starting position 0 so that it can read the data in correct order groupMetadataRecordValue.position(0) - val e = assertThrows(classOf[KafkaException], - () => GroupMetadataManager.readGroupMessageValue(groupId, groupMetadataRecordValue, time)) - assertEquals(s"Unknown group metadata version ${unsupportedVersion}", e.getMessage) + try { + GroupMetadataManager.readGroupMessageValue(groupId, groupMetadataRecordValue, time) + fail("Expected KafkaException here") + } catch { + case e: KafkaException => assertEquals(s"Unknown group metadata version ${unsupportedVersion}", e.getMessage) + case _ => fail("Expected KafkaException here") + } } @Test