kafka git commit: KAFKA-2955; Add a simple ">" prompt to console producer
Repository: kafka Updated Branches: refs/heads/trunk e108a8b4e -> b948f4327 KAFKA-2955; Add a simple ">" prompt to console producer Author: Manikumar reddy O Reviewers: Ismael Juma, Guozhang Wang, Gwen Shapira Closes #1233 from omkreddy/KAFKA-2955 Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/b948f432 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/b948f432 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/b948f432 Branch: refs/heads/trunk Commit: b948f43278e7641977c07df0069158a61fd3b443 Parents: e108a8b Author: Manikumar reddy O Authored: Tue Jan 31 09:30:29 2017 -0800 Committer: Guozhang Wang Committed: Tue Jan 31 09:30:29 2017 -0800 -- core/src/main/scala/kafka/tools/ConsoleProducer.scala | 1 + 1 file changed, 1 insertion(+) -- http://git-wip-us.apache.org/repos/asf/kafka/blob/b948f432/core/src/main/scala/kafka/tools/ConsoleProducer.scala -- diff --git a/core/src/main/scala/kafka/tools/ConsoleProducer.scala b/core/src/main/scala/kafka/tools/ConsoleProducer.scala index c6dcfce..93454d6 100644 --- a/core/src/main/scala/kafka/tools/ConsoleProducer.scala +++ b/core/src/main/scala/kafka/tools/ConsoleProducer.scala @@ -306,6 +306,7 @@ object ConsoleProducer { override def readMessage() = { lineNumber += 1 + print(">") (reader.readLine(), parseKey) match { case (null, _) => null case (line, true) =>
kafka git commit: MINOR: Logging improvements in consumer internals
Repository: kafka Updated Branches: refs/heads/trunk b948f4327 -> 5afe95964 MINOR: Logging improvements in consumer internals Author: Jason Gustafson Reviewers: Manikumar reddy O , Ewen Cheslack-Postava , Ismael Juma Closes #2469 from hachikuji/improve-consumer-logging Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/5afe9596 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/5afe9596 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/5afe9596 Branch: refs/heads/trunk Commit: 5afe959647fcad9d01365427f4b455e1586b1fd5 Parents: b948f43 Author: Jason Gustafson Authored: Tue Jan 31 12:27:00 2017 -0800 Committer: Jason Gustafson Committed: Tue Jan 31 12:27:00 2017 -0800 -- .../org/apache/kafka/clients/NetworkClient.java | 9 + .../kafka/clients/consumer/KafkaConsumer.java | 8 ++-- .../consumer/internals/AbstractCoordinator.java | 33 ++-- .../consumer/internals/ConsumerCoordinator.java | 41 .../internals/ConsumerNetworkClient.java| 9 +++-- .../clients/consumer/internals/Fetcher.java | 5 ++- .../apache/kafka/common/utils/KafkaThread.java | 9 + .../clients/consumer/KafkaConsumerTest.java | 2 - .../internals/ConsumerCoordinatorTest.java | 30 +++--- 9 files changed, 90 insertions(+), 56 deletions(-) -- http://git-wip-us.apache.org/repos/asf/kafka/blob/5afe9596/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java -- diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java index 0eb7670..3a75288 100644 --- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java +++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java @@ -427,14 +427,23 @@ public class NetworkClient implements KafkaClient { int currInflight = this.inFlightRequests.inFlightRequestCount(node.idString()); if (currInflight == 0 && this.connectionStates.isReady(node.idString())) { // if we find an established connection with no in-flight requests we can stop right away +log.trace("Found least loaded node {} connected with no in-flight requests", node); return node; } else if (!this.connectionStates.isBlackedOut(node.idString(), now) && currInflight < inflight) { // otherwise if this is the best we have found so far, record that inflight = currInflight; found = node; +} else if (log.isTraceEnabled()) { +log.trace("Removing node {} from least loaded node selection: is-blacked-out: {}, in-flight-requests: {}", +node, this.connectionStates.isBlackedOut(node.idString(), now), currInflight); } } +if (found != null) +log.trace("Found least loaded node {}", found); +else +log.trace("Least loaded node selection failed to find an available node"); + return found; } http://git-wip-us.apache.org/repos/asf/kafka/blob/5afe9596/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java -- diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index 23e7ed6..89844f6 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -78,8 +78,8 @@ import java.util.regex.Pattern; * Cross-Version Compatibility * This client can communicate with brokers that are version 0.10.0 or newer. Older or newer brokers may not support * certain features. For example, 0.10.0 brokers do not support offsetsForTimes, because this feature was added - * in version 0.10.1. You will receive an UnsupportedVersionException when invoking an API that is not available on the - * running broker version. + * in version 0.10.1. You will receive an {@link org.apache.kafka.common.errors.UnsupportedVersionException} + * when invoking an API that is not available on the running broker version. * * * Offsets and Consumer Position @@ -685,7 +685,6 @@ public class KafkaConsumer implements Consumer { metricGrpPrefix, this.time, retryBackoffMs, -new ConsumerCoordinator.DefaultOffsetCommitCallback(), config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG), config.getInt(ConsumerConfig.
kafka git commit: MINOR: Logging improvements in consumer internals
Repository: kafka Updated Branches: refs/heads/0.10.2 61024c9d2 -> a50635219 MINOR: Logging improvements in consumer internals Author: Jason Gustafson Reviewers: Manikumar reddy O , Ewen Cheslack-Postava , Ismael Juma Closes #2469 from hachikuji/improve-consumer-logging (cherry picked from commit 5afe959647fcad9d01365427f4b455e1586b1fd5) Signed-off-by: Jason Gustafson Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/a5063521 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/a5063521 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/a5063521 Branch: refs/heads/0.10.2 Commit: a5063521943f8d0f6940b18d4f0d57045aa395ae Parents: 61024c9 Author: Jason Gustafson Authored: Tue Jan 31 12:27:00 2017 -0800 Committer: Jason Gustafson Committed: Tue Jan 31 12:28:36 2017 -0800 -- .../org/apache/kafka/clients/NetworkClient.java | 9 + .../kafka/clients/consumer/KafkaConsumer.java | 8 ++-- .../consumer/internals/AbstractCoordinator.java | 33 ++-- .../consumer/internals/ConsumerCoordinator.java | 41 .../internals/ConsumerNetworkClient.java| 9 +++-- .../clients/consumer/internals/Fetcher.java | 5 ++- .../apache/kafka/common/utils/KafkaThread.java | 9 + .../clients/consumer/KafkaConsumerTest.java | 2 - .../internals/ConsumerCoordinatorTest.java | 30 +++--- 9 files changed, 90 insertions(+), 56 deletions(-) -- http://git-wip-us.apache.org/repos/asf/kafka/blob/a5063521/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java -- diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java index 0eb7670..3a75288 100644 --- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java +++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java @@ -427,14 +427,23 @@ public class NetworkClient implements KafkaClient { int currInflight = this.inFlightRequests.inFlightRequestCount(node.idString()); if (currInflight == 0 && this.connectionStates.isReady(node.idString())) { // if we find an established connection with no in-flight requests we can stop right away +log.trace("Found least loaded node {} connected with no in-flight requests", node); return node; } else if (!this.connectionStates.isBlackedOut(node.idString(), now) && currInflight < inflight) { // otherwise if this is the best we have found so far, record that inflight = currInflight; found = node; +} else if (log.isTraceEnabled()) { +log.trace("Removing node {} from least loaded node selection: is-blacked-out: {}, in-flight-requests: {}", +node, this.connectionStates.isBlackedOut(node.idString(), now), currInflight); } } +if (found != null) +log.trace("Found least loaded node {}", found); +else +log.trace("Least loaded node selection failed to find an available node"); + return found; } http://git-wip-us.apache.org/repos/asf/kafka/blob/a5063521/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java -- diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index 6064c39..ed3d607 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -78,8 +78,8 @@ import java.util.regex.Pattern; * Cross-Version Compatibility * This client can communicate with brokers that are version 0.10.0 or newer. Older or newer brokers may not support * certain features. For example, 0.10.0 brokers do not support offsetsForTimes, because this feature was added - * in version 0.10.1. You will receive an UnsupportedVersionException when invoking an API that is not available on the - * running broker version. + * in version 0.10.1. You will receive an {@link org.apache.kafka.common.errors.UnsupportedVersionException} + * when invoking an API that is not available on the running broker version. * * * Offsets and Consumer Position @@ -685,7 +685,6 @@ public class KafkaConsumer implements Consumer { metricGrpPrefix, this.time, retryBackoffMs, -new ConsumerCoordinator.DefaultOffsetCommitCallback(), config.g
kafka git commit: KAFKA-4717: Use absolute paths to files in root directory so all jars include LICENSE and NOTICE files
Repository: kafka Updated Branches: refs/heads/0.10.2 a50635219 -> 54512cbaf KAFKA-4717: Use absolute paths to files in root directory so all jars include LICENSE and NOTICE files Author: Ewen Cheslack-Postava Reviewers: Ismael Juma Closes #2473 from ewencp/kafka-4717-connect-license-and-notice-files (cherry picked from commit 82744414d50d28a7bf7b0084118b68a4411c5781) Signed-off-by: Ewen Cheslack-Postava Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/54512cba Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/54512cba Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/54512cba Branch: refs/heads/0.10.2 Commit: 54512cbaf7b624103cb1ad43bec455bbb0709c55 Parents: a506352 Author: Ewen Cheslack-Postava Authored: Tue Jan 31 16:06:10 2017 -0800 Committer: Ewen Cheslack-Postava Committed: Tue Jan 31 16:06:30 2017 -0800 -- build.gradle | 42 +- 1 file changed, 21 insertions(+), 21 deletions(-) -- http://git-wip-us.apache.org/repos/asf/kafka/blob/54512cba/build.gradle -- diff --git a/build.gradle b/build.gradle index 7a34c25..32ce14a 100644 --- a/build.gradle +++ b/build.gradle @@ -177,21 +177,21 @@ subprojects { } jar { -from '../LICENSE' -from '../NOTICE' +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" } task srcJar(type: Jar) { classifier = 'sources' -from '../LICENSE' -from '../NOTICE' +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" from sourceSets.main.allSource } task javadocJar(type: Jar, dependsOn: javadoc) { classifier 'javadoc' -from '../LICENSE' -from '../NOTICE' +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" from javadoc.destinationDir } @@ -213,15 +213,15 @@ subprojects { if(!sourceSets.test.allSource.isEmpty()) { task testJar(type: Jar) { classifier = 'test' - from '../LICENSE' - from '../NOTICE' + from "$rootDir/LICENSE" + from "$rootDir/NOTICE" from sourceSets.test.output } task testSrcJar(type: Jar, dependsOn: testJar) { classifier = 'test-sources' - from '../LICENSE' - from '../NOTICE' + from "$rootDir/LICENSE" + from "$rootDir/NOTICE" from sourceSets.test.allSource } @@ -234,8 +234,8 @@ subprojects { plugins.withType(ScalaPlugin) { task scaladocJar(type:Jar) { classifier = 'scaladoc' - from '../LICENSE' - from '../NOTICE' + from "$rootDir/LICENSE" + from "$rootDir/NOTICE" from scaladoc.destinationDir } @@ -511,7 +511,7 @@ project(':core') { ':streams:genStreamsConfigDocs'], type: Tar) { classifier = 'site-docs' compression = Compression.GZIP -from project.file("../docs") +from project.file("$rootDir/docs") into 'site-docs' duplicatesStrategy 'exclude' } @@ -519,10 +519,10 @@ project(':core') { tasks.create(name: "releaseTarGz", dependsOn: configurations.archives.artifacts, type: Tar) { into "kafka_${versions.baseScala}-${version}" compression = Compression.GZIP -from(project.file("../bin")) { into "bin/" } -from(project.file("../config")) { into "config/" } -from '../LICENSE' -from '../NOTICE' +from(project.file("$rootDir/bin")) { into "bin/" } +from(project.file("$rootDir/config")) { into "config/" } +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" from(configurations.runtime) { into("libs/") } from(configurations.archives.artifacts.files) { into("libs/") } from(project.siteDocsTar) { into("site-docs/") } @@ -606,12 +606,12 @@ project(':clients') { task determineCommitId { ext.commitId = "unknown" def takeFromHash = 16 -if (file("../.git/HEAD").exists()) { - def headRef = file("../.git/HEAD").text +if (file("$rootDir/.git/HEAD").exists()) { + def headRef = file("$rootDir/.git/HEAD").text if (headRef.contains('ref: ')) { headRef = headRef.replaceAll('ref: ', '').trim() -if (file("../.git/$headRef").exists()) { -commitId = file("../.git/$headRef").text.trim().take(takeFromHash) +if (file("$rootDir/.git/$headRef").exists()) { +commitId = file("$rootDir/.git/$headRef").text.trim().take(takeFromHash) } } else { commitId = headRef.trim().take(takeFromHash)
kafka git commit: KAFKA-4717: Use absolute paths to files in root directory so all jars include LICENSE and NOTICE files
Repository: kafka Updated Branches: refs/heads/trunk 5afe95964 -> 82744414d KAFKA-4717: Use absolute paths to files in root directory so all jars include LICENSE and NOTICE files Author: Ewen Cheslack-Postava Reviewers: Ismael Juma Closes #2473 from ewencp/kafka-4717-connect-license-and-notice-files Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/82744414 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/82744414 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/82744414 Branch: refs/heads/trunk Commit: 82744414d50d28a7bf7b0084118b68a4411c5781 Parents: 5afe959 Author: Ewen Cheslack-Postava Authored: Tue Jan 31 16:06:10 2017 -0800 Committer: Ewen Cheslack-Postava Committed: Tue Jan 31 16:06:10 2017 -0800 -- build.gradle | 42 +- 1 file changed, 21 insertions(+), 21 deletions(-) -- http://git-wip-us.apache.org/repos/asf/kafka/blob/82744414/build.gradle -- diff --git a/build.gradle b/build.gradle index 7a34c25..32ce14a 100644 --- a/build.gradle +++ b/build.gradle @@ -177,21 +177,21 @@ subprojects { } jar { -from '../LICENSE' -from '../NOTICE' +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" } task srcJar(type: Jar) { classifier = 'sources' -from '../LICENSE' -from '../NOTICE' +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" from sourceSets.main.allSource } task javadocJar(type: Jar, dependsOn: javadoc) { classifier 'javadoc' -from '../LICENSE' -from '../NOTICE' +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" from javadoc.destinationDir } @@ -213,15 +213,15 @@ subprojects { if(!sourceSets.test.allSource.isEmpty()) { task testJar(type: Jar) { classifier = 'test' - from '../LICENSE' - from '../NOTICE' + from "$rootDir/LICENSE" + from "$rootDir/NOTICE" from sourceSets.test.output } task testSrcJar(type: Jar, dependsOn: testJar) { classifier = 'test-sources' - from '../LICENSE' - from '../NOTICE' + from "$rootDir/LICENSE" + from "$rootDir/NOTICE" from sourceSets.test.allSource } @@ -234,8 +234,8 @@ subprojects { plugins.withType(ScalaPlugin) { task scaladocJar(type:Jar) { classifier = 'scaladoc' - from '../LICENSE' - from '../NOTICE' + from "$rootDir/LICENSE" + from "$rootDir/NOTICE" from scaladoc.destinationDir } @@ -511,7 +511,7 @@ project(':core') { ':streams:genStreamsConfigDocs'], type: Tar) { classifier = 'site-docs' compression = Compression.GZIP -from project.file("../docs") +from project.file("$rootDir/docs") into 'site-docs' duplicatesStrategy 'exclude' } @@ -519,10 +519,10 @@ project(':core') { tasks.create(name: "releaseTarGz", dependsOn: configurations.archives.artifacts, type: Tar) { into "kafka_${versions.baseScala}-${version}" compression = Compression.GZIP -from(project.file("../bin")) { into "bin/" } -from(project.file("../config")) { into "config/" } -from '../LICENSE' -from '../NOTICE' +from(project.file("$rootDir/bin")) { into "bin/" } +from(project.file("$rootDir/config")) { into "config/" } +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" from(configurations.runtime) { into("libs/") } from(configurations.archives.artifacts.files) { into("libs/") } from(project.siteDocsTar) { into("site-docs/") } @@ -606,12 +606,12 @@ project(':clients') { task determineCommitId { ext.commitId = "unknown" def takeFromHash = 16 -if (file("../.git/HEAD").exists()) { - def headRef = file("../.git/HEAD").text +if (file("$rootDir/.git/HEAD").exists()) { + def headRef = file("$rootDir/.git/HEAD").text if (headRef.contains('ref: ')) { headRef = headRef.replaceAll('ref: ', '').trim() -if (file("../.git/$headRef").exists()) { -commitId = file("../.git/$headRef").text.trim().take(takeFromHash) +if (file("$rootDir/.git/$headRef").exists()) { +commitId = file("$rootDir/.git/$headRef").text.trim().take(takeFromHash) } } else { commitId = headRef.trim().take(takeFromHash)
kafka git commit: KAFKA-4717: Use absolute paths to files in root directory so all jars include LICENSE and NOTICE files
Repository: kafka Updated Branches: refs/heads/0.10.1 6ba7470ed -> 850ceb74c KAFKA-4717: Use absolute paths to files in root directory so all jars include LICENSE and NOTICE files Author: Ewen Cheslack-Postava Reviewers: Ismael Juma Closes #2473 from ewencp/kafka-4717-connect-license-and-notice-files (cherry picked from commit 82744414d50d28a7bf7b0084118b68a4411c5781) Signed-off-by: Ewen Cheslack-Postava Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/850ceb74 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/850ceb74 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/850ceb74 Branch: refs/heads/0.10.1 Commit: 850ceb74ca8bafc2c22735612feb67dc9c46fc36 Parents: 6ba7470 Author: Ewen Cheslack-Postava Authored: Tue Jan 31 16:06:10 2017 -0800 Committer: Ewen Cheslack-Postava Committed: Tue Jan 31 16:07:20 2017 -0800 -- build.gradle | 42 +- 1 file changed, 21 insertions(+), 21 deletions(-) -- http://git-wip-us.apache.org/repos/asf/kafka/blob/850ceb74/build.gradle -- diff --git a/build.gradle b/build.gradle index fded48e..d221d96 100644 --- a/build.gradle +++ b/build.gradle @@ -175,21 +175,21 @@ subprojects { } jar { -from '../LICENSE' -from '../NOTICE' +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" } task srcJar(type: Jar) { classifier = 'sources' -from '../LICENSE' -from '../NOTICE' +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" from sourceSets.main.allSource } task javadocJar(type: Jar, dependsOn: javadoc) { classifier 'javadoc' -from '../LICENSE' -from '../NOTICE' +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" from javadoc.destinationDir } @@ -211,15 +211,15 @@ subprojects { if(!sourceSets.test.allSource.isEmpty()) { task testJar(type: Jar) { classifier = 'test' - from '../LICENSE' - from '../NOTICE' + from "$rootDir/LICENSE" + from "$rootDir/NOTICE" from sourceSets.test.output } task testSrcJar(type: Jar, dependsOn: testJar) { classifier = 'test-sources' - from '../LICENSE' - from '../NOTICE' + from "$rootDir/LICENSE" + from "$rootDir/NOTICE" from sourceSets.test.allSource } @@ -232,8 +232,8 @@ subprojects { plugins.withType(ScalaPlugin) { task scaladocJar(type:Jar) { classifier = 'scaladoc' - from '../LICENSE' - from '../NOTICE' + from "$rootDir/LICENSE" + from "$rootDir/NOTICE" from scaladoc.destinationDir } @@ -509,7 +509,7 @@ project(':core') { ':streams:genStreamsConfigDocs'], type: Tar) { classifier = 'site-docs' compression = Compression.GZIP -from project.file("../docs") +from project.file("$rootDir/docs") into 'site-docs' duplicatesStrategy 'exclude' } @@ -517,10 +517,10 @@ project(':core') { tasks.create(name: "releaseTarGz", dependsOn: configurations.archives.artifacts, type: Tar) { into "kafka_${versions.baseScala}-${version}" compression = Compression.GZIP -from(project.file("../bin")) { into "bin/" } -from(project.file("../config")) { into "config/" } -from '../LICENSE' -from '../NOTICE' +from(project.file("$rootDir/bin")) { into "bin/" } +from(project.file("$rootDir/config")) { into "config/" } +from "$rootDir/LICENSE" +from "$rootDir/NOTICE" from(configurations.runtime) { into("libs/") } from(configurations.archives.artifacts.files) { into("libs/") } from(project.siteDocsTar) { into("site-docs/") } @@ -602,12 +602,12 @@ project(':clients') { task determineCommitId { ext.commitId = "unknown" def takeFromHash = 16 -if (file("../.git/HEAD").exists()) { - def headRef = file("../.git/HEAD").text +if (file("$rootDir/.git/HEAD").exists()) { + def headRef = file("$rootDir/.git/HEAD").text if (headRef.contains('ref: ')) { headRef = headRef.replaceAll('ref: ', '').trim() -if (file("../.git/$headRef").exists()) { -commitId = file("../.git/$headRef").text.trim().take(takeFromHash) +if (file("$rootDir/.git/$headRef").exists()) { +commitId = file("$rootDir/.git/$headRef").text.trim().take(takeFromHash) } } else { commitId = headRef.trim().take(takeFromHash)
kafka-site git commit: Add Ewen's GPG key.
Repository: kafka-site Updated Branches: refs/heads/asf-site 60a70f379 -> 9a36603f4 Add Ewen's GPG key. Project: http://git-wip-us.apache.org/repos/asf/kafka-site/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka-site/commit/9a36603f Tree: http://git-wip-us.apache.org/repos/asf/kafka-site/tree/9a36603f Diff: http://git-wip-us.apache.org/repos/asf/kafka-site/diff/9a36603f Branch: refs/heads/asf-site Commit: 9a36603f4855890929659a123ca33a00130a1165 Parents: 60a70f3 Author: Ewen Cheslack-Postava Authored: Tue Jan 31 18:26:47 2017 -0800 Committer: Ewen Cheslack-Postava Committed: Tue Jan 31 18:26:47 2017 -0800 -- KEYS | 110 ++ 1 file changed, 110 insertions(+) -- http://git-wip-us.apache.org/repos/asf/kafka-site/blob/9a36603f/KEYS -- diff --git a/KEYS b/KEYS index b07dee0..2efca24 100644 --- a/KEYS +++ b/KEYS @@ -601,3 +601,113 @@ qoZ5VNSFiHWPXq6WyetJmAubfsJxwIWybfqJ8Dy3yivj3aCfdH4WSD2k6yqoI988 =cvL8 -END PGP PUBLIC KEY BLOCK- +pub 4096R/BE7EFC73 2015-01-14 +uid [ultimate] Ewen Cheslack-Postava +sig 3BE7EFC73 2015-01-14 Ewen Cheslack-Postava +sub 4096R/08EA97A4 2015-01-14 +sig BE7EFC73 2015-01-14 Ewen Cheslack-Postava + +pub 4096R/5234D94F 2014-08-22 +uid [ultimate] Ewen Cheslack-Postava +sig 35234D94F 2014-08-22 Ewen Cheslack-Postava +sub 4096R/964F16A8 2014-08-22 +sig 5234D94F 2014-08-22 Ewen Cheslack-Postava + +-BEGIN PGP PUBLIC KEY BLOCK- +Version: GnuPG v2 + +mQINBFS2qt4BEACksfgjdHfD2io0WZsc8jXHzx6oXE6lWnARo4hLgf1q32adgeRz +ofGV5AdiP1N5Pwus0o8PEa2UN9/rOIXavw4W0xMlWRiZlavYw6wR7zOGuFINRZJm +LFvPyIuTtAwdrsuUdOqAcw4hrz0j/V7xg7l79CalVpzO4JKnBYRtNSbg0FeVpk+0 +AEt6e3nlA/5WZLRnNwJTkSpwVkgRgmIjJh889xCtXtxRkjNoTq0H+ks12xdQyruL +nFRzQmrBCsIS9C+VCmVu8r/OBVVtwulaIYCcIxu213a23XmTmlfmHQXnvwdMPBT3 +me/2ubm6Xqon92Nc08AhxzzWp/rh9SceNWBkFzMdYnXpCIwWGRcvzJ4M+WhL5ho2 +SjaPoKj1LqFhzpeNA5F4SRPOfqvkEkGIl3L3qxXlrax2yOuJMv9sb6op01JIrQY6 +c0oQCzX5vxRdDimZfHnrcZnlbn7Wzs3N8hNwPn0+0ZGLTn//5/esNWtUkf/5cK8R +afqe0bntiNFLxvR1ysMYJf02FXRJ7rVaS02KZISrc/64bpE5GG5v6iklHrx3SJ4B +HykbITfN9n0uO2B4iivb86OnV/X5bjv52aJ3llwmsQtTyMrr+JOeccjYv+bw6SK+ +xi/t8BWlwfxa7+//5XrTgNaINE96R9EtDlMMnEU+zk0EFFQxXmTx2384DwARAQAB +tClFd2VuIENoZXNsYWNrLVBvc3RhdmEgPGV3ZW5AY29uZmx1ZW50LmlvPokCOAQT +AQIAIgUCVLaq3gIbAwYLCQgHAwIGFQgCCQoLBBYCAwECHgECF4AACgkQk07sfb5+ +/HNJIg/9GW2s+0EpKyIcvyp46Hw4y0ruOKz++JZs/vbHGWCbYQbC80gKDHxCQxvH +FgK+s4xFe9hsbD/36FBKE9Jbit581F983/KPgoD0bvX2yum6X6T/TaGFSfPkO6hw +qCbLSswLo5QJV527yrvbKe2dpL7v6E6ZOLZiP8vYJMIS30UyV40gPJp7q6yyT8gH +xEKW5aGDuCHCzRlRB+jNFUzsB6Q5MeOTVHfTvjPhHLbFhGBmU7buqxtT7EREAze5 +8I8JUFJkHtdf32a463a+2qjll6AEkWt2iBVuRilkgOnxuX/xNdb7H/6iFCT6UNCb +pMrM3RLzh0iklT5BNeZ5mMXY8FnOqBy/fFGvWedb6T9Im4ERgWrVq14X8b+uUaP9 +URRvBczXvTfSb4Lajt5BvOvgleOF7iQhzFO1byzo2HD5CArv0HJEIP7GYNDnp14g +GjQehaqmHP7/RBRDw/Hg0gy828KmfwD3K6VSWpwHOPaas84O6waxRsCKW6Ng/PpM +FWF5NfPm4sVsuK/QiCpc/hpDzpDVxLL9Z7cOG6APZKFv+g9isUTtP2sn7tDoas0Y +BvgpPz9gpASF2Y3/3wgyitzY7YO1w36jwgSqkTud8Lknc6tGOStGqclqmt4Ep/OB +43i7nbYTo12Udn37WotuDstOm8NVZWxsjXVBb7QW8Rn/fJnolZq5Ag0EVLaq3gEQ +AKN94fP55s47wo31KQhRgQH7AsLQcMh2bWcJDchIvMsF6oB582T7uHqU3AYFvJNP +FSkPc0k4XetyZTBTtbnw0bRM1UQECDvQnB4M+HpqlAM3LJce3oPzyVYZ0vxpQRcc +Y/bCyEc9j3U2BxtbHfZgCTv+keSNsE6i5dpZ92KRw/PQAyth6fDDM40DR8SiXJR2 +tIaTmXoa0JOoECPwDLQqccXmpL+5Vl4xcjrPBZ9BIiA0WBm2dvSk4Nt+8xz8FjNh +M+slsTUiI20yfKEXN9mM9pPsS7gnngZR96QJBWOBersf3L/aO1z1ywZORyYFED86 +U+I67IIHHHYVC+CsRTOLHEeSNGnYKAmuY9aQnnFkE5fkGQ46mx2eE05Od9xiciy7 +U+XKJ0Pz/yBaiSoLg5TMHo+33giHCh/ESrc+aiohjXKA6Z6cqkmQIg19sisNmZqJ +0IBNZrvNJHCxolN8VSvz1lfAyEkbYQ6L4JGdnp5Gu0DWaaJcQ2BoE/i2uD0xjNCv +o+GuEF66YI2Gm0d1oePsy+jkgrw0sZetaCccX08dJOcNklKz6nQYjFT2C43zSlzP +jkDEA9sgEabDN0SyYAnbDR+ZW0g+/B/VrWS+j+apNQAckYWAIQCzyMSbkrYJRAdO +vEXKo+rQxm/LyBIq5pMwqy1nlx87LP9h/AgtGJN6Qc0PABEBAAGJAh8EGAECAAkF +AlS2qt4CGwwACgkQk07sfb5+/HOqHw/+MhfNz5I8n7xmBvVln7sujPkiA6heWbxe +Ga+zZnqRTFFsULaMgz4dIdXwXI2Sgym3WKangiJZr3OEpLPlXSM/FS+GAruRH+/y +2CPfkqxd1jG13Q1/BWBHbtxJizWuZ+UtK+/wqWc099/4oTnkckchtZBJQ5gdGoio +swqZdj9rguXROAtW2Fmc/5K+vxgGTuE8cu4TxjA0ZEWdMaJYI6dlLg9LjPpGzs4L +jDn2dItasnsZ6irhXISYhV9WJbus2H9kYMjZxdJhY6HzkWVaUVrdtiGlWjf6fQ9N +e0X6DN9kd3YMNhiJ8JExxrB//2JPr/3B5nc/Z0zKmwn1znUz04WWKQ1QM9Irszrr +5fWAFy0LgsiT0922hH0f9puss5+irQUSIXDaW2iaCdNszZiTZxNwJDYu8m0pj7eP +KgkKutptC/1on/mE9IeOkQ5acj2XfgKvljMP9VBX5PrRoRjJ/PMxinZPbzAHZy20 +7TYMbJH5dsqFIeh66jycn2IDKElvk7A9Wk98XbaxEFq8KhxblG+glV+kz2NWm3x6 +wJ/WFx5uXvKXuXrir92HGgzC6hy+v7m1DPCBlqM3+qrXTcyhbqeC52pjj4AAFnaF +G9VPAuAnKuZSCkxGuiEgxxjDdikTkYmxbqpel+m0T6DLLgNX58M2yURAIaDajLW/ +gozBq5tLHNiZAg0EU/a4MQEQAMohxV5xUHkknnH4h3xr6tOczInCW3xObdUlsZh9 +mLhiU4+0kzQtNbjKK2qgNgmQzL1p+VYuQ2XQJmhxynXAcc9sFag7kOnzrVvC+HGb +gZRr79YrtGqEFwU/a3YommEVAFdVvBe0HM+24pbBMoiHV1OaabnSmAYX2Fmpuh+t +xo7FvPZL6/X
kafka git commit: KAFKA-4677: Avoid unnecessary task movement across threads during rebalance
Repository: kafka Updated Branches: refs/heads/trunk 82744414d -> 0b48ea1c8 KAFKA-4677: Avoid unnecessary task movement across threads during rebalance Makes task assignment more sticky by preferring to assign tasks to clients that had previously had the task as active task. If there are no clients with the task previously active, then search for a standby. Finally falling back to the least loaded client. Author: Damian Guy Reviewers: Matthias J. Sax, Guozhang Wang Closes #2429 from dguy/kafka-4677 Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/0b48ea1c Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/0b48ea1c Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/0b48ea1c Branch: refs/heads/trunk Commit: 0b48ea1c81f22465cf32a19c012e0fb3c849afcc Parents: 8274441 Author: Damian Guy Authored: Tue Jan 31 20:16:47 2017 -0800 Committer: Guozhang Wang Committed: Tue Jan 31 20:16:47 2017 -0800 -- .../internals/StreamPartitionAssignor.java | 34 +- .../processor/internals/StreamThread.java | 12 +- .../internals/assignment/ClientState.java | 116 - .../assignment/StickyTaskAssignor.java | 283 ++ .../internals/assignment/TaskAssignor.java | 208 +--- .../kstream/internals/KTableAggregateTest.java | 5 + .../internals/StreamPartitionAssignorTest.java | 20 +- .../processor/internals/StreamThreadTest.java | 4 +- .../internals/assignment/ClientStateTest.java | 151 ++ .../assignment/StickyTaskAssignorTest.java | 515 +++ .../internals/assignment/TaskAssignorTest.java | 312 --- 11 files changed, 1085 insertions(+), 575 deletions(-) -- http://git-wip-us.apache.org/repos/asf/kafka/blob/0b48ea1c/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java -- diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java index 1ad6dbc..e17d96b 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java @@ -33,8 +33,8 @@ import org.apache.kafka.streams.processor.TaskId; import org.apache.kafka.streams.processor.TopologyBuilder; import org.apache.kafka.streams.processor.internals.assignment.AssignmentInfo; import org.apache.kafka.streams.processor.internals.assignment.ClientState; +import org.apache.kafka.streams.processor.internals.assignment.StickyTaskAssignor; import org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfo; -import org.apache.kafka.streams.processor.internals.assignment.TaskAssignor; import org.apache.kafka.streams.state.HostInfo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -105,13 +105,10 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable } void addConsumer(final String consumerMemberId, final SubscriptionInfo info) { - consumers.add(consumerMemberId); - -state.prevActiveTasks.addAll(info.prevTasks); -state.prevAssignedTasks.addAll(info.prevTasks); -state.prevAssignedTasks.addAll(info.standbyTasks); -state.capacity = state.capacity + 1d; +state.addPreviousActiveTasks(info.prevTasks); +state.addPreviousStandbyTasks(info.standbyTasks); +state.incrementCapacity(); } @Override @@ -228,10 +225,10 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable // 2. Task ids of previously running tasks // 3. Task ids of valid local states on the client's state directory. -Set prevTasks = streamThread.prevTasks(); +final Set previousActiveTasks = streamThread.prevActiveTasks(); Set standbyTasks = streamThread.cachedTasks(); -standbyTasks.removeAll(prevTasks); -SubscriptionInfo data = new SubscriptionInfo(streamThread.processId, prevTasks, standbyTasks, this.userEndPoint); +standbyTasks.removeAll(previousActiveTasks); +SubscriptionInfo data = new SubscriptionInfo(streamThread.processId, previousActiveTasks, standbyTasks, this.userEndPoint); if (streamThread.builder.sourceTopicPattern() != null) { SubscriptionUpdates subscriptionUpdates = new SubscriptionUpdates(); @@ -461,7 +458,8 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable log.debug("stream-thread [{}] Assigning tasks {} to clients {} with number of replicas {