diff --git a/.github/workflows/build-and-test-on-pr-events.yml b/.github/workflows/build-and-test-on-pr-events.yml new file mode 100644 index 0000000000000..ef0799288f409 --- /dev/null +++ b/.github/workflows/build-and-test-on-pr-events.yml @@ -0,0 +1,46 @@ +# 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 + +name: PR Builds + +# Run this workflow every time a new pull request is created in the repository +on: + pull_request: + types: [opened, reopened, synchronize] + +jobs: + build: + # Name the Job + name: Build Pull Request and run all unit tests + # Set the type of machine to run on + runs-on: ubuntu-latest + steps: + # Checks out a copy of your repository on the ubuntu-latest machine + - name: Checkout code + uses: actions/checkout@v2 + with: + ref: ${{ github.event.pull_request.head.sha }} + # bring in all history because the gradle versions plugin needs to "walk back" to the closest ancestor tag + fetch-depth: 0 + - name: Set up JDK 1.8 + uses: actions/setup-java@v1 + with: + java-version: 1.8 + - name: Build with Gradle and run all unit tests + # exclude the streams test and connect test + # Set maxTestRetries for flaky tests + run: ./gradlew -PmaxTestRetries=3 cleanTest rat checkstyleMain checkstyleTest :clients:unitTest :core:unitTest --no-daemon -PxmlFindBugsReport=true -PtestLoggingEvents=started,passed,skipped,failed diff --git a/.github/workflows/build-and-upload-archives-upon-creating-tags.yml b/.github/workflows/build-and-upload-archives-upon-creating-tags.yml new file mode 100644 index 0000000000000..544e0d6a5cecd --- /dev/null +++ b/.github/workflows/build-and-upload-archives-upon-creating-tags.yml @@ -0,0 +1,59 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License + +name: Release build on tags + +# Run this workflow every time a tag is created/pushed +on: + push: + tags: + - '*' + +jobs: + build: + # Name the Job + name: Build tagged commit and upload an archive + # Set the type of machine to run on + runs-on: ubuntu-latest + steps: + # Checks out a copy of your repository on the ubuntu-latest machine + - name: Checkout code + uses: actions/checkout@v2 + with: + ref: ${{ github.event.pull_request.head.sha }} + # bring in all history because the gradle versions plugin needs to "walk back" to the closest ancestor tag + fetch-depth: 0 + - name: Set up JDK 1.8 + uses: actions/setup-java@v1 + with: + java-version: 1.8 + - name: Set up release version env variable + run: | + echo "RELEASE_VERSION=${GITHUB_REF#refs/*/}" >> $GITHUB_ENV + - name: Print the release version + run: | + echo "Release version (tag name): ${{ env.RELEASE_VERSION }}" + - name: Build with Gradle and run all tests + # exclude the streams test and connect test + # Set maxTestRetries for flaky tests + run: ./gradlew -PmaxTestRetries=3 cleanTest rat checkstyleMain checkstyleTest :clients:test :core:test --no-daemon -PxmlFindBugsReport=true -PtestLoggingEvents=started,passed,skipped,failed + - name: Upload archive + env: + JFROG_USERNAME: ${{ secrets.JFROG_USERNAME }} + JFROG_API_KEY: ${{ secrets.JFROG_API_KEY }} + run: | + ./gradlew -Pversion=${{ env.RELEASE_VERSION }} :clients:uploadArchives :core:uploadArchives --no-daemon diff --git a/.github/workflows/build-on-push-to-release-branch.yml b/.github/workflows/build-on-push-to-release-branch.yml new file mode 100644 index 0000000000000..05b56ef063807 --- /dev/null +++ b/.github/workflows/build-on-push-to-release-branch.yml @@ -0,0 +1,49 @@ +# 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 + +name: Commit pushed build + +# Run this workflow every time a new pull request is created in the repository +on: + push: + branches: + - 3.0-li + - 3.0-li-dev2 + +jobs: + build: + # Name the Job + name: Build pushed commits with integration tests + # Set the type of machine to run on + runs-on: ubuntu-latest + steps: + # Checks out a copy of your repository on the ubuntu-latest machine + - name: Checkout code + uses: actions/checkout@v2 + with: + ref: ${{ github.event.pull_request.head.sha }} + # bring in all history because the gradle versions plugin needs to "walk back" to the closest ancestor tag + fetch-depth: 0 + - name: Set up JDK 1.8 + uses: actions/setup-java@v1 + with: + java-version: 1.8 + - name: Build with Gradle and run all unit tests + # exclude the streams test and connect test + # Run integration tests when pushed to branch + # Set maxTestRetries for flaky tests + run: ./gradlew -PmaxTestRetries=3 cleanTest rat checkstyleMain checkstyleTest :clients:test :core:test --no-daemon -PxmlFindBugsReport=true -PtestLoggingEvents=started,passed,skipped,failed diff --git a/README.md b/README.md index 0fbfe890e2018..56876b307b1db 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,59 @@ +testtest LinkedIn Branch of Apache Kafka +================= + +This is the version of Kafka running at LinkedIn. + +Kafka was born at LinkedIn. We run thousands of brokers to deliver trillions of +messages per day. We run a slightly modified version of Apache Kafka trunk. +This branch contains the LinkedIn Kafka release. + +This branch is made up of: + +* Apache Kafka trunk (upstream) up to some branch point, see *-li* branch name for base version, you'll be able to get the exact commit from git +* Cherry-picked commits from upstream after branch point +* Patches that are on their way upstream but we have deployed internally in the meantime +* Patches that are of no interest to upstream + +We are making this branch available for people interested. We will be +documenting the changes in the near future with some more detailed explanations +in the [LinkedIn Engineering Blog](https://engineering.linkedin.com/blog). + +If you are interested in learning more, we invite you to our [Streaming +Meetup](https://www.meetup.com/Stream-Processing-Meetup-LinkedIn/) where we +discuss streaming technologies like [Kafka](http://kafka.apache.org) and +[Samza](http://samza.apache.org). + +You are encouraged to check out other Kafka projects from LinkedIn: + +* [Cruise Control](https://github.com/linkedin/cruise-control) +* [Li-Apache-Kafka-Clients](https://github.com/linkedin/li-apache-kafka-clients) +* [Burrow](https://github.com/linkedin/Burrow) +* [Kafka Monitor](https://github.com/linkedin/kafka-monitor) + +### CI ### +We are currently using Github Actions as the CI framework, and the testing results can be found [here](https://github.com/linkedin/kafka/actions). +To publish a release, go to [the release page](https://github.com/linkedin/kafka/releases) and manually create a new release. +Once the release tag is created, a test job will be triggered to run the necessary tests. And once the test passes, the artifacts +will be published to [the bintray hosting LinkedIn projects](https://dl.bintray.com/linkedin/maven/com/linkedin/kafka/kafka_2.12/). + +Currently we've configured the CI flow to run only unit tests for 'clients' and 'core' when a pull request is created or updated: + ./gradlew :clients:unitTest :core:unitTest +In contrast, all tests for `clients' and `core' are run when creating a release, which may be significantly longer than running the unit tests: + ./gradlew :clients:test :core:test +The reason for this mixed approach is to get faster feedback from CI during code reviews +and still gain the more through test coverage when publishing a release. + +### Contributing ### + +At this moment we are not accepting external contributions directly. Please +contribute to [Apache Kafka](http://kafka.apache.org). + +For security issues with this branch please review +[LinkedIn Security +Guidelines](https://www.linkedin.com/help/linkedin/answer/62924/security-vulnerabilities?lang=en). +General Kafka issues should be communicated via the Kafka community. + + Apache Kafka ================= See our [web site](https://kafka.apache.org) for details on the project. @@ -132,20 +188,29 @@ The `eclipse` task has been configured to use `${project_dir}/build_eclipse` as build directory (`${project_dir}/bin`) clashes with Kafka's scripts directory and we don't use Gradle's build directory to avoid known issues with this configuration. -### Publishing the jar for all version of Scala and for all projects to maven ### +### Publishing the jar for all projects to maven ### The recommended command is: - ./gradlewAll publish + ./gradlew -Pversion= publish For backwards compatibility, the following also works: - ./gradlewAll uploadArchives + ./gradlew -Pversion= uploadArchives + +By default, this command will publish artifacts to a JFrog repository named "kafka" under an account specified by the `JFROG_USERNAME` environment variable; +and the `JFROG_API_KEY` environment variable is used for the API key for that account. + +If you want to publish for all supported Scala version, change `./gradlew` to `./gradlewAll`. -Please note for this to work you should create/update `${GRADLE_USER_HOME}/gradle.properties` (typically, `~/.gradle/gradle.properties`) and assign the following variables +If you want to override this to use a different maven repository, you should create/update `${GRADLE_USER_HOME}/gradle.properties` (typically, `~/.gradle/gradle.properties`) +and assign the following variables mavenUrl= mavenUsername= mavenPassword= + +Signing is disabled by default. If you need signing, please set the following variables in `gradle.properties` as well: + signing.keyId= signing.password= signing.secretKeyRingFile= diff --git a/build.gradle b/build.gradle index 7627dc13ae669..39b8f1344be19 100644 --- a/build.gradle +++ b/build.gradle @@ -114,9 +114,13 @@ ext { skipSigning = project.hasProperty('skipSigning') && skipSigning.toBoolean() shouldSign = !skipSigning && !version.endsWith("SNAPSHOT") - mavenUrl = project.hasProperty('mavenUrl') ? project.mavenUrl : '' - mavenUsername = project.hasProperty('mavenUsername') ? project.mavenUsername : '' - mavenPassword = project.hasProperty('mavenPassword') ? project.mavenPassword : '' + jfrogUsername = System.getenv('JFROG_USERNAME') + jfrogApiKey = System.getenv('JFROG_API_KEY') + + // By default, publish to JFrog. + mavenUrl = project.hasProperty('mavenUrl') ? project.mavenUrl : jfrogRepoUrl + mavenUsername = project.hasProperty('mavenUsername') ? project.mavenUsername : jfrogUsername + mavenPassword = project.hasProperty('mavenPassword') ? project.mavenPassword : jfrogApiKey userShowStandardStreams = project.hasProperty("showStandardStreams") ? showStandardStreams : null @@ -260,8 +264,8 @@ subprojects { artifactId = archivesBaseName pom { - name = 'Apache Kafka' - url = 'https://kafka.apache.org' + name = 'LinkedIn fork of Apache Kafka' + url = 'https://github.com/linkedin/kafka' licenses { license { name = 'The Apache License, Version 2.0' diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml index 67f257a93c276..dcbd5fcf55981 100644 --- a/checkstyle/import-control.xml +++ b/checkstyle/import-control.xml @@ -196,6 +196,7 @@ + diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java index 8cf0100068751..60753639d9af9 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java @@ -40,6 +40,7 @@ import org.apache.kafka.common.message.LeaveGroupResponseData.MemberResponse; import org.apache.kafka.common.message.SyncGroupRequestData; import org.apache.kafka.common.metrics.Measurable; +import org.apache.kafka.common.metrics.MetricConfig; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.stats.Avg; @@ -1306,6 +1307,21 @@ public GroupCoordinatorMetrics(Metrics metrics, String metricGrpPrefix) { this.metricGrpName, "The number of seconds since the last coordinator heartbeat was sent"), lastHeartbeat); + + //HOTFIX - extra liveliness-related metrics + + Measurable lastHeartbeatReceived = + new Measurable() { + public double measure(MetricConfig config, long now) { + return TimeUnit.SECONDS.convert(now - heartbeat.lastHeartbeatReceive(), TimeUnit.MILLISECONDS); + } + }; + metrics.addMetric(metrics.metricName("last-heartbeat-received-seconds-ago", + this.metricGrpName, + "The number of seconds since the last successful controller heartbeat was received"), + lastHeartbeatReceived); + + //end HOTFIX } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java index 67ad51e0849a7..1dbc5aad85fa0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java @@ -17,6 +17,7 @@ package org.apache.kafka.clients.consumer.internals; import org.apache.kafka.clients.GroupRebalanceConfig; +import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.CommitFailedException; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerGroupMetadata; @@ -47,6 +48,7 @@ import org.apache.kafka.common.message.OffsetCommitRequestData; import org.apache.kafka.common.message.OffsetCommitResponseData; import org.apache.kafka.common.metrics.Measurable; +import org.apache.kafka.common.metrics.MetricConfig; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.stats.Avg; @@ -112,6 +114,9 @@ public final class ConsumerCoordinator extends AbstractCoordinator { private ConsumerGroupMetadata groupMetadata; private final boolean throwOnFetchStableOffsetsUnsupported; + private volatile long prevPollTime = Long.MIN_VALUE; //volatile for metrics + + // hold onto request&future for committed offset requests to enable async calls. private PendingCommittedOffsetRequest pendingCommittedOffsetRequest = null; @@ -467,6 +472,11 @@ void maybeUpdateSubscriptionMetadata() { * @return true iff the operation succeeded */ public boolean poll(Timer timer, boolean waitForJoinGroup) { + long currentTime = time.milliseconds(); + if (prevPollTime > Long.MIN_VALUE) { + sensors.pollInterval.record(currentTime - prevPollTime); + } + prevPollTime = currentTime; maybeUpdateSubscriptionMetadata(); invokeCompletedOffsetCommitCallbacks(); @@ -1408,6 +1418,7 @@ private class ConsumerCoordinatorMetrics { private final Sensor revokeCallbackSensor; private final Sensor assignCallbackSensor; private final Sensor loseCallbackSensor; + private final Sensor pollInterval; private ConsumerCoordinatorMetrics(Metrics metrics, String metricGrpPrefix) { this.metricGrpName = metricGrpPrefix + "-coordinator-metrics"; @@ -1449,6 +1460,30 @@ private ConsumerCoordinatorMetrics(Metrics metrics, String metricGrpPrefix) { metrics.addMetric(metrics.metricName("assigned-partitions", this.metricGrpName, "The number of partitions currently assigned to this consumer"), numParts); + + //HOTFIX - extra liveliness-related metrics + + this.pollInterval = metrics.sensor("poll-interval"); + this.pollInterval.add(metrics.metricName("poll-interval-avg", + this.metricGrpName, + "The average time between subsequent poll calls"), new Avg()); + this.pollInterval.add(metrics.metricName("poll-interval-max", + this.metricGrpName, + "The max time between subsequent poll calls"), new Max()); + this.pollInterval.add(createMeter(metrics, metricGrpName, "poll", "poll calls")); + + Measurable lastHeartbeat = + new Measurable() { + public double measure(MetricConfig config, long now) { + return TimeUnit.SECONDS.convert(now - prevPollTime, TimeUnit.MILLISECONDS); + } + }; + metrics.addMetric(metrics.metricName("last-poll-seconds-ago", + this.metricGrpName, + "The number of seconds since the last poll call"), + lastHeartbeat); + + //end HOTFIX } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Heartbeat.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Heartbeat.java index dfb9f85144d2d..55e552fd0698d 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Heartbeat.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Heartbeat.java @@ -37,6 +37,7 @@ public final class Heartbeat { private volatile long lastHeartbeatSend = 0L; private volatile boolean heartbeatInFlight = false; + private volatile long lastHeartbeatReceive; public Heartbeat(GroupRebalanceConfig config, Time time) { @@ -54,6 +55,7 @@ public Heartbeat(GroupRebalanceConfig config, } private void update(long now) { + lastHeartbeatReceive = now; heartbeatTimer.update(now); sessionTimer.update(now); pollTimer.update(now); @@ -97,6 +99,10 @@ boolean shouldHeartbeat(long now) { update(now); return heartbeatTimer.isExpired(); } + + public long lastHeartbeatReceive() { + return lastHeartbeatReceive; + } long lastHeartbeatSend() { return this.lastHeartbeatSend; diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java index ef8a9cc4fd68b..49398f61f3894 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java @@ -1123,11 +1123,28 @@ private void ensureValidRecordSize(int size) { */ @Override public void flush() { + flush(Integer.MAX_VALUE, TimeUnit.MILLISECONDS); + } + + /** + * This method waits up to timeout for the producer to send out all the buffered records. + * @param timeout The maximum time to wait for producer to complete. The value should be non-negative. + * @param unit The time unit for the timeout + * @throws TimeoutException If producer fail to finish in time + * @throws InterruptException If the thread is interrupted while blocked + * @throws IllegalArgumentException If the timeout is negative. + */ + @Override + public void flush(long timeout, TimeUnit unit) { + if (timeout < 0) + throw new IllegalArgumentException("The timeout cannot be negative."); log.trace("Flushing accumulated records in producer."); - this.accumulator.beginFlush(); - this.sender.wakeup(); try { - this.accumulator.awaitFlushCompletion(); + this.accumulator.beginFlush(); + this.sender.wakeup(); + this.accumulator.awaitFlushCompletion(unit.toMillis(timeout)); + } catch (TimeoutException e) { + throw e; } catch (InterruptedException e) { throw new InterruptException("Flush interrupted.", e); } diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java index 4fd540dceaa8a..aaec569f2c1a6 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java @@ -17,6 +17,7 @@ package org.apache.kafka.clients.producer; import org.apache.kafka.clients.consumer.ConsumerGroupMetadata; +import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.producer.internals.DefaultPartitioner; import org.apache.kafka.clients.producer.internals.FutureRecordMetadata; @@ -274,7 +275,7 @@ private void verifyTransactionInFlight() { /** * Adds the record to the list of sent records. The {@link RecordMetadata} returned will be immediately satisfied. - * + * * @see #history() */ @Override @@ -347,7 +348,7 @@ private long nextOffset(TopicPartition tp) { } } - public synchronized void flush() { + public synchronized void flush(long timeout, TimeUnit unit) { verifyProducerState(); if (this.flushException != null) { @@ -358,6 +359,10 @@ public synchronized void flush() { completeNext(); } + public synchronized void flush() { + flush(Long.MAX_VALUE, null); + } + public List partitionsFor(String topic) { if (this.partitionsForException != null) { throw this.partitionsForException; diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/Producer.java b/clients/src/main/java/org/apache/kafka/clients/producer/Producer.java index 4f3e9ec0d7281..0f276e1a7d9ff 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/Producer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/Producer.java @@ -29,6 +29,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; /** * The interface for the {@link KafkaProducer} @@ -86,6 +87,11 @@ void sendOffsetsToTransaction(Map offsets, void flush(); /** + * See {@link KafkaProducer#flush(long, TimeUnit)} + */ + void flush(long timeout, TimeUnit unit); + + /** * See {@link KafkaProducer#partitionsFor(String)} */ List partitionsFor(String topic); diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java index ee84c7c168a4a..b73b73a6e1d37 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java @@ -19,6 +19,9 @@ import java.nio.ByteBuffer; import java.util.ArrayDeque; import java.util.Deque; +import java.util.Iterator; +import java.util.LinkedHashSet; +import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; @@ -30,6 +33,8 @@ import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.stats.Meter; import org.apache.kafka.common.utils.Time; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** @@ -44,12 +49,14 @@ */ public class BufferPool { + private static final Logger log = LoggerFactory.getLogger(BufferPool.class); + static final String WAIT_TIME_SENSOR_NAME = "bufferpool-wait-time"; private final long totalMemory; private final int poolableSize; private final ReentrantLock lock; - private final Deque free; + private final Set free; private final Deque waiters; /** Total available memory is the sum of nonPooledAvailableMemory and the number of byte buffers in free * poolableSize. */ private long nonPooledAvailableMemory; @@ -57,6 +64,7 @@ public class BufferPool { private final Time time; private final Sensor waitTime; private boolean closed; + private long nextOvermemoryWarn; /** * Create a new buffer pool @@ -70,7 +78,7 @@ public class BufferPool { public BufferPool(long memory, int poolableSize, Metrics metrics, Time time, String metricGrpName) { this.poolableSize = poolableSize; this.lock = new ReentrantLock(); - this.free = new ArrayDeque<>(); + this.free = new LinkedHashSet<>(); this.waiters = new ArrayDeque<>(); this.totalMemory = memory; this.nonPooledAvailableMemory = memory; @@ -91,6 +99,7 @@ public BufferPool(long memory, int poolableSize, Metrics metrics, Time time, Str this.waitTime.add(new Meter(TimeUnit.NANOSECONDS, rateMetricName, totalMetricName)); this.closed = false; + this.nextOvermemoryWarn = 0; } /** @@ -122,7 +131,7 @@ public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedEx try { // check if we have a free buffer of the right size pooled if (size == poolableSize && !this.free.isEmpty()) - return this.free.pollFirst(); + return popBuffer(); // now check if the request is immediately satisfiable with the // memory on hand or if we need to block @@ -167,7 +176,7 @@ public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedEx // otherwise allocate memory if (accumulated == 0 && size == this.poolableSize && !this.free.isEmpty()) { // just grab a buffer from the free list - buffer = this.free.pollFirst(); + buffer = popBuffer(); accumulated = size; } else { // we'll need to allocate memory, but we may only get @@ -244,25 +253,36 @@ protected ByteBuffer allocateByteBuffer(int size) { */ private void freeUp(int size) { while (!this.free.isEmpty() && this.nonPooledAvailableMemory < size) - this.nonPooledAvailableMemory += this.free.pollLast().capacity(); + this.nonPooledAvailableMemory += popBuffer().capacity(); } /** * Return buffers to the pool. If they are of the poolable size add them to the free list, otherwise just mark the * memory as free. * - * @param buffer The buffer to return + * @param buffer The buffer to return to the pool. * @param size The size of the buffer to mark as deallocated, note that this may be smaller than buffer.capacity - * since the buffer may re-allocate itself during in-place compression + * since the buffer may re-allocate itself during in-place compression */ public void deallocate(ByteBuffer buffer, int size) { lock.lock(); try { - if (size == this.poolableSize && size == buffer.capacity()) { - buffer.clear(); - this.free.add(buffer); + long availableMemory = availableMemoryUnlocked(); + if (availableMemory + size > this.totalMemory && nextOvermemoryWarn < this.time.milliseconds()) { + //Don't flood logs. + log.error("Detected an attempt to bring available memory " + availableMemory + " to " + + (availableMemory + size) + " which is higher than totalMemory " + this.totalMemory + ".", new Exception()); + this.nextOvermemoryWarn = this.time.milliseconds() + TimeUnit.HOURS.toMillis(1); + } + if (buffer.hasArray() && buffer.array().length == this.poolableSize && + this.poolableSize + availableMemory <= this.totalMemory) { + if (!this.free.add(buffer.array()) && this.nextOvermemoryWarn < time.milliseconds()) { + log.error("Detected an attempt to double deallocate the same buffer.", new Exception()); + this.nextOvermemoryWarn = this.time.milliseconds() + TimeUnit.HOURS.toMillis(1); + } } else { - this.nonPooledAvailableMemory += size; + long freeMem = Math.min(Math.max(buffer.capacity(), size), this.totalMemory - availableMemory); + this.nonPooledAvailableMemory += freeMem; } Condition moreMem = this.waiters.peekFirst(); if (moreMem != null) @@ -282,12 +302,16 @@ public void deallocate(ByteBuffer buffer) { public long availableMemory() { lock.lock(); try { - return this.nonPooledAvailableMemory + freeSize() * (long) this.poolableSize; + return availableMemoryUnlocked(); } finally { lock.unlock(); } } + private long availableMemoryUnlocked() { + return this.nonPooledAvailableMemory + freeSize() * (long) this.poolableSize; + } + // Protected for testing. protected int freeSize() { return this.free.size(); @@ -350,4 +374,11 @@ public void close() { this.lock.unlock(); } } + + private ByteBuffer popBuffer() { + Iterator it = free.iterator(); + byte[] array = it.next(); + it.remove(); + return ByteBuffer.wrap(array); + } } diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java index 24a80b9fe592b..afc17d0b6a602 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java @@ -29,6 +29,7 @@ import java.util.Set; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.ApiVersions; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.common.utils.ProducerIdAndEpoch; @@ -38,6 +39,7 @@ import org.apache.kafka.common.Node; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.header.Header; import org.apache.kafka.common.metrics.Measurable; @@ -712,16 +714,26 @@ private boolean appendsInProgress() { } /** - * Mark all partitions as ready to send and block until the send is complete + * Mark all partitions as ready to send and block until the send is complete or time expires */ - public void awaitFlushCompletion() throws InterruptedException { + public void awaitFlushCompletion(long timeoutMs) throws InterruptedException { try { // Obtain a copy of all of the incomplete ProduceRequestResult(s) at the time of the flush. // We must be careful not to hold a reference to the ProduceBatch(s) so that garbage // collection can occur on the contents. // The sender will remove ProducerBatch(s) from the original incomplete collection. - for (ProduceRequestResult result : this.incomplete.requestResults()) - result.await(); + Long expireMs = System.currentTimeMillis() + timeoutMs; + for (ProduceRequestResult result : this.incomplete.requestResults()) { + Long currentMs = System.currentTimeMillis(); + if (currentMs > expireMs) { + throw new TimeoutException("Failed to flush accumulated records within" + timeoutMs + "milliseconds."); + } + + boolean completed = result.await(Math.max(expireMs - currentMs, 0), TimeUnit.MILLISECONDS); + if (!completed) { + throw new TimeoutException("Failed to flush accumulated records within" + timeoutMs + "milliseconds."); + } + } } finally { this.flushesInProgress.decrementAndGet(); } diff --git a/clients/src/main/java/org/apache/kafka/common/utils/AppInfoParser.java b/clients/src/main/java/org/apache/kafka/common/utils/AppInfoParser.java index 19f98d1b652aa..dd50e32eec646 100644 --- a/clients/src/main/java/org/apache/kafka/common/utils/AppInfoParser.java +++ b/clients/src/main/java/org/apache/kafka/common/utils/AppInfoParser.java @@ -61,8 +61,9 @@ public static synchronized void registerAppInfo(String prefix, String id, Metric try { ObjectName name = new ObjectName(prefix + ":type=app-info,id=" + Sanitizer.jmxSanitize(id)); AppInfo mBean = new AppInfo(nowMs); - ManagementFactory.getPlatformMBeanServer().registerMBean(mBean, name); - + if (!ManagementFactory.getPlatformMBeanServer().isRegistered(name)) { + ManagementFactory.getPlatformMBeanServer().registerMBean(mBean, name); + } registerMetrics(metrics, mBean); // prefix will be added later by JmxReporter } catch (JMException e) { log.warn("Error registering AppInfo mbean", e); diff --git a/clients/src/main/java/org/apache/kafka/common/utils/PoisonPill.java b/clients/src/main/java/org/apache/kafka/common/utils/PoisonPill.java new file mode 100644 index 0000000000000..b98853762ed60 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/utils/PoisonPill.java @@ -0,0 +1,112 @@ +/* + * 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.common.utils; + +import com.sun.management.HotSpotDiagnosticMXBean; +import java.io.File; +import java.lang.management.ManagementFactory; +import java.nio.file.Files; +import java.nio.file.StandardCopyOption; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import javax.management.MBeanServer; + + +public class PoisonPill { + + public static void die() { + die(null, -1, 1); + } + + public static void die(File heapDumpFolder, final long maxWaitForDump) { + die(heapDumpFolder, maxWaitForDump, 1); + } + + public static void die(File heapDumpFolder, final long maxWaitForDump, int haltStatusCode) { + try { + if (maxWaitForDump > 0 && heapDumpFolder != null) { + grabHeapDump(heapDumpFolder, maxWaitForDump, haltStatusCode); + } + } catch (Exception e) { + System.err.println("unable to complete heap dump"); + e.printStackTrace(System.err); + System.err.flush(); + } finally { + Runtime.getRuntime().halt(haltStatusCode); + } + } + + private static void grabHeapDump(File heapDumpFolder, final long maxWait, final int haltStatusCode) throws Exception { + + //set up a watchdog background thread that will halt in ~maxWait regardless of whether or not + //we succeed in taking a heap dump (since we dont know when it'll ever complete) + final CountDownLatch latch = new CountDownLatch(1); + Thread watchdog = new Thread(new Runnable() { + @Override + public void run() { + try { + latch.countDown(); + Thread.sleep(maxWait); + //at this point ~maxWait has passed since the call to die(). + //if the heap dump process completed successfully die() would + //have called halt() and we wouldnt be here (99.99%) + System.err.println("heap dump (probably) did not complete within timeout. halting."); + System.err.flush(); + } catch (Exception e) { + System.err.println("watchdog caught exception"); + e.printStackTrace(System.err); + System.err.flush(); + } finally { + Runtime.getRuntime().halt(haltStatusCode); + } + } + }); + watchdog.setDaemon(true); + watchdog.setName("clark the death watchdog"); + watchdog.start(); + + //make sure the watchdog is up and running before we go off attempting to dump + if (!latch.await(maxWait, TimeUnit.MILLISECONDS)) { + System.err.println("unable to start watchdog within timeout. will not proceed with dump"); + System.err.flush(); + return; + } + + System.err.println("dumping heap to " + heapDumpFolder.getCanonicalPath()); + System.err.flush(); + + //we dump into dump.inprogress and atomically rename it to be dump.complete + //(overwriting any previous such file). this attempts to guarantee there are + //at most 2 (potentially large) dump files at any point in time. + + File inProgress = new File(heapDumpFolder, "dump.inprogress"); + File complete = new File(heapDumpFolder, "dump.complete"); + if (inProgress.exists() && !inProgress.delete()) { + System.err.println("unable to delete existing dump file. will not proceed with dump"); + System.err.flush(); + return; + } + + MBeanServer server = ManagementFactory.getPlatformMBeanServer(); + HotSpotDiagnosticMXBean diagnosticMBean = + ManagementFactory.newPlatformMXBeanProxy(server, "com.sun.management:type=HotSpotDiagnostic", + HotSpotDiagnosticMXBean.class); + diagnosticMBean.dumpHeap(inProgress.getCanonicalPath(), false /* disable only live - dump all objects */); + Files.move(inProgress.toPath(), complete.toPath(), StandardCopyOption.ATOMIC_MOVE, + StandardCopyOption.REPLACE_EXISTING); + } +} diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java index eaf35f9b8d025..2be6c797ec72f 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java @@ -355,6 +355,23 @@ protected ByteBuffer allocateByteBuffer(int size) { assertEquals(bufferPool.availableMemory(), 1024); } + @Test + public void overDeallocate() { + BufferPool bufferPool = new BufferPool(1024, 512, metrics, time, metricGroup); + bufferPool.deallocate(ByteBuffer.allocate(512)); + assertEquals(bufferPool.availableMemory(), bufferPool.totalMemory()); + } + + @Test + public void dedupeDeallocations() throws Exception { + BufferPool bufferPool = new BufferPool(1024, 512, metrics, time, metricGroup); + ByteBuffer bbuf = bufferPool.allocate(512, 1); + ByteBuffer shallowCopy = bbuf.slice(); + bufferPool.deallocate(bbuf); + bufferPool.deallocate(shallowCopy); + assertEquals(bufferPool.totalMemory(), bufferPool.availableMemory()); + } + public static class StressTestThread extends Thread { private final int iterations; private final BufferPool pool; diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java index 06ed1ce1f1242..6e5c2f9d020da 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java @@ -26,6 +26,7 @@ import org.apache.kafka.common.Node; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.protocol.ApiKeys; @@ -401,7 +402,7 @@ public void testFlush() throws Exception { accum.deallocate(batch); // should be complete with no unsent records. - accum.awaitFlushCompletion(); + accum.awaitFlushCompletion(Integer.MAX_VALUE); assertFalse(accum.hasUndrained()); assertFalse(accum.hasIncomplete()); } @@ -427,13 +428,30 @@ public void testAwaitFlushComplete() throws Exception { assertTrue(accum.flushInProgress()); delayedInterrupt(Thread.currentThread(), 1000L); try { - accum.awaitFlushCompletion(); + accum.awaitFlushCompletion(Integer.MAX_VALUE); fail("awaitFlushCompletion should throw InterruptException"); } catch (InterruptedException e) { assertFalse(accum.flushInProgress(), "flushInProgress count should be decremented even if thread is interrupted"); } } + @Test + public void testAwaitFlushTimeout() throws Exception { + RecordAccumulator accum = createTestRecordAccumulator( + 4 * 1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 64 * 1024, CompressionType.NONE, Integer.MAX_VALUE); + accum.append(new TopicPartition(topic, 0), 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, false, System.currentTimeMillis()); + + accum.beginFlush(); + assertTrue(accum.flushInProgress()); + try { + accum.awaitFlushCompletion(100); + fail("testAwaitFlushTimeout should throw TimeoutException"); + } catch (TimeoutException e) { + assertFalse(accum.flushInProgress(), "flushInProgress count should be decremented even if flush timeout expires"); + } + } + + @Test public void testAbortIncompleteBatches() throws Exception { int lingerMs = Integer.MAX_VALUE; diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala index 2f10710ddfef5..49711e49bef35 100755 --- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala +++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala @@ -41,7 +41,7 @@ import org.apache.kafka.common.{KafkaException, Node, Reconfigurable, TopicParti import scala.jdk.CollectionConverters._ import scala.collection.mutable.HashMap -import scala.collection.{Seq, Set, mutable} +import scala.collection.{Seq, Map, Set, mutable} object ControllerChannelManager { val QueueSizeMetricName = "QueueSize" @@ -58,6 +58,7 @@ class ControllerChannelManager(controllerContext: ControllerContext, protected val brokerStateInfo = new HashMap[Int, ControllerBrokerStateInfo] private val brokerLock = new Object + val brokerResponseSensors: mutable.Map[ApiKeys, BrokerResponseTimeStats] = mutable.HashMap.empty this.logIdent = "[Channel manager on controller " + config.brokerId + "]: " newGauge("TotalQueueSize", @@ -72,12 +73,27 @@ class ControllerChannelManager(controllerContext: ControllerContext, brokerLock synchronized { brokerStateInfo.foreach(brokerState => startRequestSendThread(brokerState._1)) } + initBrokerResponseSensors() } def shutdown() = { brokerLock synchronized { brokerStateInfo.values.toList.foreach(removeExistingBroker) } + removeBrokerResponseSensors() + } + + def initBrokerResponseSensors(): Unit = { + Array(ApiKeys.STOP_REPLICA, ApiKeys.LEADER_AND_ISR, ApiKeys.UPDATE_METADATA).foreach { k: ApiKeys => + brokerResponseSensors.put(k, new BrokerResponseTimeStats(k)) + } + } + + def removeBrokerResponseSensors(): Unit = { + brokerResponseSensors.keySet.foreach { k: ApiKeys => + brokerResponseSensors(k).removeMetrics() + brokerResponseSensors.remove(k) + } } def sendRequest(brokerId: Int, request: AbstractControlRequest.Builder[_ <: AbstractControlRequest], @@ -173,7 +189,7 @@ class ControllerChannelManager(controllerContext: ControllerContext, ) val requestThread = new RequestSendThread(config.brokerId, controllerContext, messageQueue, networkClient, - brokerNode, config, time, requestRateAndQueueTimeMetrics, stateChangeLogger, threadName) + brokerNode, config, time, requestRateAndQueueTimeMetrics, stateChangeLogger, threadName, this) requestThread.setDaemon(false) val queueSizeGauge = newGauge(QueueSizeMetricName, () => messageQueue.size, brokerMetricTags(broker.id)) @@ -221,7 +237,8 @@ class RequestSendThread(val controllerId: Int, val time: Time, val requestRateAndQueueTimeMetrics: Timer, val stateChangeLogger: StateChangeLogger, - name: String) + name: String, + val controllerChannelManager: ControllerChannelManager) extends ShutdownableThread(name = name) { logIdent = s"[RequestSendThread controllerId=$controllerId] " @@ -233,7 +250,9 @@ class RequestSendThread(val controllerId: Int, def backoff(): Unit = pause(100, TimeUnit.MILLISECONDS) val QueueItem(apiKey, requestBuilder, callback, enqueueTimeMs) = queue.take() - requestRateAndQueueTimeMetrics.update(time.milliseconds() - enqueueTimeMs, TimeUnit.MILLISECONDS) + var queueTimeMs = time.milliseconds() - enqueueTimeMs + var remoteTimeMs: Long = 0 + requestRateAndQueueTimeMetrics.update(queueTimeMs, TimeUnit.MILLISECONDS) var clientResponse: ClientResponse = null try { @@ -251,6 +270,7 @@ class RequestSendThread(val controllerId: Int, time.milliseconds(), true) clientResponse = NetworkClientUtils.sendAndReceive(networkClient, clientRequest, time) isSendSuccessful = true + remoteTimeMs = time.milliseconds() - enqueueTimeMs - queueTimeMs } } catch { case e: Throwable => // if the send was not successful, reconnect to broker and resend the message @@ -276,6 +296,7 @@ class RequestSendThread(val controllerId: Int, if (callback != null) { callback(response) } + controllerChannelManager.brokerResponseSensors(api).update(queueTimeMs, remoteTimeMs) } } catch { case e: Throwable => @@ -681,3 +702,22 @@ case class ControllerBrokerStateInfo(networkClient: NetworkClient, requestRateAndTimeMetrics: Timer, reconfigurableChannelBuilder: Option[Reconfigurable]) + +class BrokerResponseTimeStats(val key: ApiKeys) extends KafkaMetricsGroup { + // Records time for request waits on local send thread queue + val brokerRequestQueueTime = newHistogram("brokerRequestQueueTimeMs", true, responseTimeTags) + // Records time for controller to send request and receive response + val brokerRequestRemoteTime = newHistogram("brokerRequestRemoteTimeMs", true, responseTimeTags) + + def responseTimeTags = Map("request" -> key.toString) + + def update(queueTime: Long, remoteTime: Long): Unit = { + brokerRequestQueueTime.update(queueTime) + brokerRequestRemoteTime.update(remoteTime) + } + + def removeMetrics(): Unit = { + removeMetric("brokerRequestQueueTimeMs", responseTimeTags) + removeMetric("brokerRequestRemoteTimeMs", responseTimeTags) + } +} diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala index 50b00e3fca344..23d35ba19ecb4 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala @@ -91,6 +91,9 @@ class GroupCoordinator(val brokerId: Int, props.put(LogConfig.CleanupPolicyProp, LogConfig.Compact) props.put(LogConfig.SegmentBytesProp, offsetConfig.offsetsTopicSegmentBytes.toString) props.put(LogConfig.CompressionTypeProp, ProducerCompressionCodec.name) + props.put(LogConfig.MaxMessageBytesProp, offsetConfig.offsetsTopicMaxMessageBytes.toString) + props.put(LogConfig.MinInSyncReplicasProp, offsetConfig.offsetsTopicMinInSyncReplicas.toString) + props.put(LogConfig.MinCompactionLagMsProp, offsetConfig.offsetsTopicMinCompactionLagMs.toString) props } @@ -1642,7 +1645,10 @@ object GroupCoordinator { offsetsTopicReplicationFactor = config.offsetsTopicReplicationFactor, offsetsTopicCompressionCodec = config.offsetsTopicCompressionCodec, offsetCommitTimeoutMs = config.offsetCommitTimeoutMs, - offsetCommitRequiredAcks = config.offsetCommitRequiredAcks + offsetCommitRequiredAcks = config.offsetCommitRequiredAcks, + offsetsTopicMaxMessageBytes = config.offsetsTopicMaxMessageBytes, + offsetsTopicMinInSyncReplicas = config.offsetsTopicMinInSyncReplicas, + offsetsTopicMinCompactionLagMs = config.offsetsTopicMinCompactionLagMs ) def apply(config: KafkaConfig, diff --git a/core/src/main/scala/kafka/coordinator/group/OffsetConfig.scala b/core/src/main/scala/kafka/coordinator/group/OffsetConfig.scala index 55ec590852cd0..f4a22d4da6ddf 100644 --- a/core/src/main/scala/kafka/coordinator/group/OffsetConfig.scala +++ b/core/src/main/scala/kafka/coordinator/group/OffsetConfig.scala @@ -36,6 +36,9 @@ import kafka.message.{CompressionCodec, NoCompressionCodec} * commit or this timeout is reached. (Similar to the producer request timeout.) * @param offsetCommitRequiredAcks The required acks before the commit can be accepted. In general, the default (-1) * should not be overridden. + * @param offsetsTopicMaxMessageBytes The maximum record batch size for the offset commit topic + * @param offsetsTopicMinInSyncReplicas The minimum number of replicas that must acknowledged a write for the write to be considered successful + * @param offsetsTopicMinCompactionLagMs The minimum time a message will stay un-compacted in the log */ case class OffsetConfig(maxMetadataSize: Int = OffsetConfig.DefaultMaxMetadataSize, loadBufferSize: Int = OffsetConfig.DefaultLoadBufferSize, @@ -46,7 +49,10 @@ case class OffsetConfig(maxMetadataSize: Int = OffsetConfig.DefaultMaxMetadataSi offsetsTopicReplicationFactor: Short = OffsetConfig.DefaultOffsetsTopicReplicationFactor, offsetsTopicCompressionCodec: CompressionCodec = OffsetConfig.DefaultOffsetsTopicCompressionCodec, offsetCommitTimeoutMs: Int = OffsetConfig.DefaultOffsetCommitTimeoutMs, - offsetCommitRequiredAcks: Short = OffsetConfig.DefaultOffsetCommitRequiredAcks) + offsetCommitRequiredAcks: Short = OffsetConfig.DefaultOffsetCommitRequiredAcks, + offsetsTopicMaxMessageBytes: Int = OffsetConfig.DefaultOffsetsTopicMaxMessageBytes, + offsetsTopicMinInSyncReplicas: Int = OffsetConfig.DefaultOffsetsTopicMinInSyncReplicas, + offsetsTopicMinCompactionLagMs: Long = OffsetConfig.DefaultOffsetsTopicMinCompactionLagMs) object OffsetConfig { val DefaultMaxMetadataSize = 4096 @@ -59,4 +65,7 @@ object OffsetConfig { val DefaultOffsetsTopicCompressionCodec = NoCompressionCodec val DefaultOffsetCommitTimeoutMs = 5000 val DefaultOffsetCommitRequiredAcks = (-1).toShort + val DefaultOffsetsTopicMaxMessageBytes = 20 * 1024 * 1024 + val DefaultOffsetsTopicMinInSyncReplicas = 1 + val DefaultOffsetsTopicMinCompactionLagMs = 0L } \ No newline at end of file diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala b/core/src/main/scala/kafka/log/LogCleaner.scala index 1f1d776f83b86..7f6e537ad90d6 100644 --- a/core/src/main/scala/kafka/log/LogCleaner.scala +++ b/core/src/main/scala/kafka/log/LogCleaner.scala @@ -144,6 +144,8 @@ class LogCleaner(initialConfig: CleanerConfig, newGauge("DeadThreadCount", () => deadThreadCount) private[log] def deadThreadCount: Int = cleaners.count(_.isThreadFailed) + /* a metric to track the number of cleaner threads alive */ + newGauge("live-cleaner-thread-count", () => cleaners.count(_.asInstanceOf[Thread].isAlive)) /** * Start the background cleaning diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala index 5e456b065e057..67e44ecc0a2ad 100644 --- a/core/src/main/scala/kafka/network/RequestChannel.scala +++ b/core/src/main/scala/kafka/network/RequestChannel.scala @@ -344,6 +344,13 @@ class RequestChannel(val queueSize: Int, val requestQueueSizeMetricName = metricNamePrefix.concat(RequestQueueSizeMetric) val responseQueueSizeMetricName = metricNamePrefix.concat(ResponseQueueSizeMetric) + // Set this to Long.Maxvalue so that KafkaHealthCheck will not shutdown broker if it + // reads lastDequeueTimeMs before lastDequeueTimeMs is updated by any KafkaRequestHandler thread. + @volatile var lastDequeueTimeMs = Long.MaxValue + // This metric can help user select a suitable threshold for requestMaxLocalTimeMs so that broker can shutdown itself only when it + // is stuck or too slow. A suggested value of requestMaxLocalTimeMs could be twice the 999'th percentile of the RequestDequeuePollIntervalMs. + private val requestDequeuePollIntervalMs = newHistogram("RequestDequeuePollIntervalMs") + newGauge(requestQueueSizeMetricName, () => requestQueue.size) newGauge(responseQueueSizeMetricName, () => { @@ -446,12 +453,20 @@ class RequestChannel(val queueSize: Int, } /** Get the next request or block until specified time has elapsed */ - def receiveRequest(timeout: Long): RequestChannel.BaseRequest = + def receiveRequest(timeout: Long): RequestChannel.BaseRequest = { + val curTime = time.milliseconds + requestDequeuePollIntervalMs.update(curTime - lastDequeueTimeMs) + lastDequeueTimeMs = curTime requestQueue.poll(timeout, TimeUnit.MILLISECONDS) + } /** Get the next request or block until there is one */ - def receiveRequest(): RequestChannel.BaseRequest = + def receiveRequest(): RequestChannel.BaseRequest = { + val curTime = time.milliseconds + requestDequeuePollIntervalMs.update(curTime - lastDequeueTimeMs) + lastDequeueTimeMs = curTime requestQueue.take() + } def updateErrorMetrics(apiKey: ApiKeys, errors: collection.Map[Errors, Integer]): Unit = { errors.forKeyValue { (error, count) => diff --git a/core/src/main/scala/kafka/server/ClientQuotaManager.scala b/core/src/main/scala/kafka/server/ClientQuotaManager.scala index 1e6523f524eef..0b47039508720 100644 --- a/core/src/main/scala/kafka/server/ClientQuotaManager.scala +++ b/core/src/main/scala/kafka/server/ClientQuotaManager.scala @@ -23,7 +23,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock import kafka.network.RequestChannel import kafka.network.RequestChannel._ import kafka.server.ClientQuotaManager._ -import kafka.utils.{Logging, QuotaUtils, ShutdownableThread} +import kafka.utils.{KafkaScheduler, Logging, QuotaUtils, ShutdownableThread} import org.apache.kafka.common.{Cluster, MetricName} import org.apache.kafka.common.metrics._ import org.apache.kafka.common.metrics.Metrics @@ -184,6 +184,7 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig, private val metrics: Metrics, private val quotaType: QuotaType, private val time: Time, + private val schedulerOpt: Option[KafkaScheduler], private val threadNamePrefix: String, private val clientQuotaCallback: Option[ClientQuotaCallback] = None) extends Logging { @@ -198,7 +199,7 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig, case None => QuotaTypes.NoQuotas } - private val delayQueueSensor = metrics.sensor(quotaType.toString + "-delayQueue") + private val delayQueueSensor = metrics.sensor(quotaType.toString + "-delayQueueSize") delayQueueSensor.add(metrics.metricName("queue-size", quotaType.toString, "Tracks the size of the delay queue"), new CumulativeSum()) @@ -207,8 +208,18 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig, start() // Use start method to keep spotbugs happy private def start(): Unit = { throttledChannelReaper.start() + schedulerOpt match { + case Some(scheduler) => + scheduler.schedule("quota-metrics-logger-%s".format(quotaType), logQuotaMetrics, 60, 60, TimeUnit.SECONDS) + case _ => + } } + private val throttledRequestCountSensor = metrics.sensor(quotaType.toString() + "-throttledRequestCount") + throttledRequestCountSensor.add(metrics.metricName("throttle-count", + quotaType.toString, + "Tracks the number of requests that have been throttled"), new CumulativeSum()) + /** * Reaper thread that triggers channel unmute callbacks on all throttled channels * @param delayQueue DelayQueue to dequeue from @@ -227,6 +238,19 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig, } } + def logQuotaMetrics(): Unit = { + val metricsMap = metrics.metrics().asScala + metricsMap.foreach { + case (metricName: MetricName, kafkaMetric: KafkaMetric) => + if (metricName.group().equals(quotaType.toString) && + (metricName.name().equals("byte-rate") || metricName.name().equals("throttle-time") || + metricName.name().equals("request-time")) && + (metricName.tags().containsKey("client-id") || metricName.tags().containsKey("user"))) { + info("Metric name (" + metricName + ") has value (" + kafkaMetric.metricValue() + ")") + } + } + } + /** * Returns true if any quotas are enabled for this quota manager. This is used * to determine if quota related metrics should be created. @@ -338,6 +362,7 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig, val throttledChannel = new ThrottledChannel(time, throttleTimeMs, throttleCallback) delayQueue.add(throttledChannel) delayQueueSensor.record() + throttledRequestCountSensor.record() debug("Channel throttled for sensor (%s). Delay time: (%d)".format(clientSensors.quotaSensor.name(), throttleTimeMs)) } } diff --git a/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala b/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala index 2ceaab9c9afdf..31c745fe1bc2e 100644 --- a/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala +++ b/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala @@ -19,7 +19,7 @@ package kafka.server import java.util.concurrent.TimeUnit import kafka.network.RequestChannel -import kafka.utils.QuotaUtils +import kafka.utils.{QuotaUtils, KafkaScheduler} import org.apache.kafka.common.MetricName import org.apache.kafka.common.metrics._ import org.apache.kafka.common.utils.Time @@ -37,9 +37,10 @@ object ClientRequestQuotaManager { class ClientRequestQuotaManager(private val config: ClientQuotaManagerConfig, private val metrics: Metrics, private val time: Time, + private val schedulerOpt: Option[KafkaScheduler], private val threadNamePrefix: String, private val quotaCallback: Option[ClientQuotaCallback]) - extends ClientQuotaManager(config, metrics, QuotaType.Request, time, threadNamePrefix, quotaCallback) { + extends ClientQuotaManager(config, metrics, QuotaType.Request, time, schedulerOpt, threadNamePrefix, quotaCallback) { private val maxThrottleTimeMs = TimeUnit.SECONDS.toMillis(this.config.quotaWindowSizeSeconds) private val exemptMetricName = metrics.metricName("exempt-request-time", diff --git a/core/src/main/scala/kafka/server/ControllerMutationQuotaManager.scala b/core/src/main/scala/kafka/server/ControllerMutationQuotaManager.scala index f011a6b36632d..b1d6e041a3c30 100644 --- a/core/src/main/scala/kafka/server/ControllerMutationQuotaManager.scala +++ b/core/src/main/scala/kafka/server/ControllerMutationQuotaManager.scala @@ -166,7 +166,7 @@ class ControllerMutationQuotaManager(private val config: ClientQuotaManagerConfi private val time: Time, private val threadNamePrefix: String, private val quotaCallback: Option[ClientQuotaCallback]) - extends ClientQuotaManager(config, metrics, QuotaType.ControllerMutation, time, threadNamePrefix, quotaCallback) { + extends ClientQuotaManager(config, metrics, QuotaType.ControllerMutation, time, None, threadNamePrefix, quotaCallback) { override protected def clientQuotaMetricName(quotaMetricTags: Map[String, String]): MetricName = { metrics.metricName("tokens", QuotaType.ControllerMutation.toString, diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index c556d7ab813b4..24db9de9914f5 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -17,6 +17,7 @@ package kafka.server +import java.io.File import java.util import java.util.{Collections, Locale, Properties} import kafka.api.{ApiVersion, ApiVersionValidator, KAFKA_0_10_0_IV1, KAFKA_2_1_IV0, KAFKA_2_7_IV0, KAFKA_2_8_IV0, KAFKA_3_0_IV1} @@ -94,6 +95,7 @@ object Defaults { val SocketSendBufferBytes: Int = 100 * 1024 val SocketReceiveBufferBytes: Int = 100 * 1024 val SocketRequestMaxBytes: Int = 100 * 1024 * 1024 + val RequestMaxLocalTimeMs = Long.MaxValue val MaxConnectionsPerIp: Int = Int.MaxValue val MaxConnectionsPerIpOverrides: String = "" val MaxConnections: Int = Int.MaxValue @@ -104,6 +106,9 @@ object Defaults { val ConnectionSetupTimeoutMaxMs = CommonClientConfigs.DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS val FailedAuthenticationDelayMs = 100 + val HeapDumpFolder = "." + val HeapDumpTimeout = 30000 + /** ********* Log Configuration ***********/ val NumPartitions = 1 val LogDir = "/tmp/kafka-logs" @@ -194,6 +199,9 @@ object Defaults { val OffsetsRetentionCheckIntervalMs: Long = OffsetConfig.DefaultOffsetsRetentionCheckIntervalMs val OffsetCommitTimeoutMs = OffsetConfig.DefaultOffsetCommitTimeoutMs val OffsetCommitRequiredAcks = OffsetConfig.DefaultOffsetCommitRequiredAcks + val OffsetsTopicMaxMessageBytes = OffsetConfig.DefaultOffsetsTopicMaxMessageBytes + val OffsetsTopicMinInSyncReplicas = OffsetConfig.DefaultOffsetsTopicMinInSyncReplicas + val OffsetsTopicMinCompactionLagMs = OffsetConfig.DefaultOffsetsTopicMinCompactionLagMs /** ********* Transaction management configuration ***********/ val TransactionalIdExpirationMs = TransactionStateManager.DefaultTransactionalIdExpirationMs @@ -392,6 +400,8 @@ object KafkaConfig { val MetadataMaxRetentionMillisProp = "metadata.max.retention.ms" val QuorumVotersProp = RaftConfig.QUORUM_VOTERS_CONFIG + val HeapDumpFolderProp = "heap.dump.folder" + val HeapDumpTimeoutProp = "heap.dump.timeout" /************* Authorizer Configuration ***********/ val AuthorizerClassNameProp = "authorizer.class.name" /** ********* Socket Server Configuration ***********/ @@ -401,6 +411,7 @@ object KafkaConfig { val ControlPlaneListenerNameProp = "control.plane.listener.name" val SocketSendBufferBytesProp = "socket.send.buffer.bytes" val SocketReceiveBufferBytesProp = "socket.receive.buffer.bytes" + val RequestMaxLocalTimeMsProp = "request.max.local.time.ms" val SocketRequestMaxBytesProp = "socket.request.max.bytes" val MaxConnectionsPerIpProp = "max.connections.per.ip" val MaxConnectionsPerIpOverridesProp = "max.connections.per.ip.overrides" @@ -507,6 +518,9 @@ object KafkaConfig { val OffsetsRetentionCheckIntervalMsProp = "offsets.retention.check.interval.ms" val OffsetCommitTimeoutMsProp = "offsets.commit.timeout.ms" val OffsetCommitRequiredAcksProp = "offsets.commit.required.acks" + val OffsetsTopicMaxMessageBytesProp = "offsets.topic.max.message.bytes" + val OffsetsTopicMinInSyncReplicasProp = "offsets.topic.min.insync.replicas" + val OffsetsTopicMinCompactionLagMsProp = "offsets.topic.min.compaction.lag.ms" /** ********* Transaction management configuration ***********/ val TransactionalIdExpirationMsProp = "transactional.id.expiration.ms" val TransactionsMaxTimeoutMsProp = "transaction.max.timeout.ms" @@ -696,6 +710,8 @@ object KafkaConfig { val MetadataMaxRetentionMillisDoc = "The number of milliseconds to keep a metadata log file or snapshot before " + "deleting it. Since at least one snapshot must exist before any logs can be deleted, this is a soft limit." + val HeapDumpFolderDoc = "The Folder under which heap dumps will be written by the watchdog" + val HeapDumpTimeoutDoc = "The max amount of time (in millis) to wait for heap dump to complete before halting regardless" /************* Authorizer Configuration ***********/ val AuthorizerClassNameDoc = s"The fully qualified name of a class that implements s${classOf[Authorizer].getName}" + " interface, which is used by the broker for authorization." @@ -743,6 +759,9 @@ object KafkaConfig { val SocketSendBufferBytesDoc = "The SO_SNDBUF buffer of the socket server sockets. If the value is -1, the OS default will be used." val SocketReceiveBufferBytesDoc = "The SO_RCVBUF buffer of the socket server sockets. If the value is -1, the OS default will be used." + val RequestMaxLocalTimeMsDoc = "The maximum allowable request local processing time. If a request's local processing " + + "takes longer than this time the broker will kill itself as violating this timeout is a symptom of a more serious broker zombie state." + + " It is useful to observe the RequestDequeuePollIntervalMs metric to find a suitable setting for this configuration." val SocketRequestMaxBytesDoc = "The maximum number of bytes in a socket request" val MaxConnectionsPerIpDoc = "The maximum number of connections we allow from each ip address. This can be set to 0 if there are overrides " + s"configured using $MaxConnectionsPerIpOverridesProp property. New connections from the ip address are dropped if the limit is reached." @@ -901,6 +920,9 @@ object KafkaConfig { val OffsetCommitTimeoutMsDoc = "Offset commit will be delayed until all replicas for the offsets topic receive the commit " + "or this timeout is reached. This is similar to the producer request timeout." val OffsetCommitRequiredAcksDoc = "The required acks before the commit can be accepted. In general, the default (-1) should not be overridden" + val OffsetsTopicMaxMessageBytesDoc = "Overriden " + MessageMaxBytesProp + " config for the consumer_offset topic." + val OffsetsTopicMinInSyncReplicasDoc = "Overridden " + MinInSyncReplicasProp + " config for the consumer_offset topic." + val OffsetsTopicMinCompactionLagMsDoc = "Overridden " + LogCleanerMinCompactionLagMsProp + " config for the consumer_offset topic." /** ********* Transaction management configuration ***********/ val TransactionalIdExpirationMsDoc = "The time in ms that the transaction coordinator will wait without receiving any transaction status updates " + "for the current transaction before expiring its transactional id. This setting also influences producer id expiration - producer ids are expired " + @@ -1090,6 +1112,8 @@ object KafkaConfig { .define(MetadataLogSegmentMillisProp, LONG, Defaults.LogRollHours * 60 * 60 * 1000L, null, HIGH, MetadataLogSegmentMillisDoc) .define(MetadataMaxRetentionBytesProp, LONG, Defaults.LogRetentionBytes, null, HIGH, MetadataMaxRetentionBytesDoc) .define(MetadataMaxRetentionMillisProp, LONG, Defaults.LogRetentionHours * 60 * 60 * 1000L, null, HIGH, MetadataMaxRetentionMillisDoc) + .define(HeapDumpFolderProp, STRING, Defaults.HeapDumpFolder, LOW, HeapDumpFolderDoc) + .define(HeapDumpTimeoutProp, LONG, Defaults.HeapDumpTimeout, LOW, HeapDumpTimeoutDoc) /************* Authorizer Configuration ***********/ .define(AuthorizerClassNameProp, STRING, Defaults.AuthorizerClassName, LOW, AuthorizerClassNameDoc) @@ -1100,6 +1124,7 @@ object KafkaConfig { .define(ListenerSecurityProtocolMapProp, STRING, Defaults.ListenerSecurityProtocolMap, LOW, ListenerSecurityProtocolMapDoc) .define(ControlPlaneListenerNameProp, STRING, null, HIGH, controlPlaneListenerNameDoc) .define(SocketSendBufferBytesProp, INT, Defaults.SocketSendBufferBytes, HIGH, SocketSendBufferBytesDoc) + .define(RequestMaxLocalTimeMsProp, LONG, Defaults.RequestMaxLocalTimeMs, atLeast(1), MEDIUM, RequestMaxLocalTimeMsDoc) .define(SocketReceiveBufferBytesProp, INT, Defaults.SocketReceiveBufferBytes, HIGH, SocketReceiveBufferBytesDoc) .define(SocketRequestMaxBytesProp, INT, Defaults.SocketRequestMaxBytes, atLeast(1), HIGH, SocketRequestMaxBytesDoc) .define(MaxConnectionsPerIpProp, INT, Defaults.MaxConnectionsPerIp, atLeast(0), MEDIUM, MaxConnectionsPerIpDoc) @@ -1210,6 +1235,9 @@ object KafkaConfig { .define(OffsetCommitRequiredAcksProp, SHORT, Defaults.OffsetCommitRequiredAcks, HIGH, OffsetCommitRequiredAcksDoc) .define(DeleteTopicEnableProp, BOOLEAN, Defaults.DeleteTopicEnable, HIGH, DeleteTopicEnableDoc) .define(CompressionTypeProp, STRING, Defaults.CompressionType, HIGH, CompressionTypeDoc) + .define(OffsetsTopicMaxMessageBytesProp, INT, Defaults.OffsetsTopicMaxMessageBytes, atLeast(0), HIGH, OffsetsTopicMaxMessageBytesDoc) + .define(OffsetsTopicMinInSyncReplicasProp, INT, Defaults.OffsetsTopicMinInSyncReplicas, atLeast(1), HIGH, OffsetsTopicMinInSyncReplicasDoc) + .define(OffsetsTopicMinCompactionLagMsProp, LONG, Defaults.OffsetsTopicMinCompactionLagMs, atLeast(0), HIGH, OffsetsTopicMinCompactionLagMsDoc) /** ********* Transaction management configuration ***********/ .define(TransactionalIdExpirationMsProp, INT, Defaults.TransactionalIdExpirationMs, atLeast(1), HIGH, TransactionalIdExpirationMsDoc) @@ -1565,6 +1593,8 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO val requestTimeoutMs = getInt(KafkaConfig.RequestTimeoutMsProp) val connectionSetupTimeoutMs = getLong(KafkaConfig.ConnectionSetupTimeoutMsProp) val connectionSetupTimeoutMaxMs = getLong(KafkaConfig.ConnectionSetupTimeoutMaxMsProp) + val heapDumpFolder = new File(getString(KafkaConfig.HeapDumpFolderProp)) + val heapDumpTimeout = getLong(KafkaConfig.HeapDumpTimeoutProp) def getNumReplicaAlterLogDirsThreads: Int = { val numThreads: Integer = Option(getInt(KafkaConfig.NumReplicaAlterLogDirsThreadsProp)).getOrElse(logDirs.size) @@ -1588,6 +1618,7 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO val socketSendBufferBytes = getInt(KafkaConfig.SocketSendBufferBytesProp) val socketReceiveBufferBytes = getInt(KafkaConfig.SocketReceiveBufferBytesProp) val socketRequestMaxBytes = getInt(KafkaConfig.SocketRequestMaxBytesProp) + val requestMaxLocalTimeMs = getLong(KafkaConfig.RequestMaxLocalTimeMsProp) val maxConnectionsPerIp = getInt(KafkaConfig.MaxConnectionsPerIpProp) val maxConnectionsPerIpOverrides: Map[String, Int] = getMap(KafkaConfig.MaxConnectionsPerIpOverridesProp, getString(KafkaConfig.MaxConnectionsPerIpOverridesProp)).map { case (k, v) => (k, v.toInt)} @@ -1700,6 +1731,9 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO val offsetCommitRequiredAcks = getShort(KafkaConfig.OffsetCommitRequiredAcksProp) val offsetsTopicSegmentBytes = getInt(KafkaConfig.OffsetsTopicSegmentBytesProp) val offsetsTopicCompressionCodec = Option(getInt(KafkaConfig.OffsetsTopicCompressionCodecProp)).map(value => CompressionCodec.getCompressionCodec(value)).orNull + val offsetsTopicMaxMessageBytes = getInt(KafkaConfig.OffsetsTopicMaxMessageBytesProp) + val offsetsTopicMinInSyncReplicas = getInt(KafkaConfig.OffsetsTopicMinInSyncReplicasProp) + val offsetsTopicMinCompactionLagMs = getLong(KafkaConfig.OffsetsTopicMinCompactionLagMsProp) /** ********* Transaction management configuration ***********/ val transactionalIdExpirationMs = getInt(KafkaConfig.TransactionalIdExpirationMsProp) @@ -1919,7 +1953,7 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO if (voterIds.isEmpty) { throw new ConfigException(s"If using ${KafkaConfig.ProcessRolesProp}, ${KafkaConfig.QuorumVotersProp} must contain a parseable set of voters.") } else if (processRoles.contains(ControllerRole)) { - // Ensure that controllers use their node.id as a voter in controller.quorum.voters + // Ensure that controllers use their node.id as a voter in controller.quorum.voters require(voterIds.contains(nodeId), s"If ${KafkaConfig.ProcessRolesProp} contains the 'controller' role, the node id $nodeId must be included in the set of voters ${KafkaConfig.QuorumVotersProp}=$voterIds") } else { // Ensure that the broker's node.id is not an id in controller.quorum.voters @@ -2015,7 +2049,7 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO val principalBuilderClass = getClass(KafkaConfig.PrincipalBuilderClassProp) require(principalBuilderClass != null, s"${KafkaConfig.PrincipalBuilderClassProp} must be non-null") - require(classOf[KafkaPrincipalSerde].isAssignableFrom(principalBuilderClass), + require(classOf[KafkaPrincipalSerde].isAssignableFrom(principalBuilderClass), s"${KafkaConfig.PrincipalBuilderClassProp} must implement KafkaPrincipalSerde") } } diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index 96576cd350243..5701b4f081dea 100755 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -45,7 +45,7 @@ import org.apache.kafka.common.requests.{ControlledShutdownRequest, ControlledSh import org.apache.kafka.common.security.scram.internals.ScramMechanism import org.apache.kafka.common.security.token.delegation.internals.DelegationTokenCache import org.apache.kafka.common.security.{JaasContext, JaasUtils} -import org.apache.kafka.common.utils.{AppInfoParser, LogContext, Time, Utils} +import org.apache.kafka.common.utils.{AppInfoParser, LogContext, Time, Utils, PoisonPill} import org.apache.kafka.common.{Endpoint, Node} import org.apache.kafka.metadata.BrokerState import org.apache.kafka.server.authorizer.Authorizer @@ -162,6 +162,16 @@ class KafkaServer( val featureCache: FinalizedFeatureCache = new FinalizedFeatureCache(brokerFeatures) override def brokerState: BrokerState = _brokerState + private var healthCheckScheduler: KafkaScheduler = null + + private def haltIfNotHealthy(): Unit = { + // This relies on io-thread to receive request from RequestChannel with 300 ms timeout, so that lastDequeueTimeMs + // will keep increasing even if there is no incoming request + if (time.milliseconds - socketServer.dataPlaneRequestChannel.lastDequeueTimeMs > config.requestMaxLocalTimeMs) { + fatal(s"It has been more than ${config.requestMaxLocalTimeMs} ms since the last time any io-thread reads from RequestChannel. Shutdown broker now.") + PoisonPill.die(config.heapDumpFolder, config.heapDumpTimeout) + } + } def clusterId: String = _clusterId @@ -243,7 +253,7 @@ class KafkaServer( /* register broker metrics */ _brokerTopicStats = new BrokerTopicStats - quotaManagers = QuotaFactory.instantiate(config, metrics, time, threadNamePrefix.getOrElse("")) + quotaManagers = QuotaFactory.instantiate(config, metrics, time, Some(kafkaScheduler), threadNamePrefix.getOrElse("")) KafkaBroker.notifyClusterListeners(clusterId, kafkaMetricsReporters ++ metrics.reporters.asScala) logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size) @@ -318,6 +328,13 @@ class KafkaServer( val brokerInfo = createBrokerInfo val brokerEpoch = zkClient.registerBroker(brokerInfo) + healthCheckScheduler = new KafkaScheduler(threads = 1, threadNamePrefix = "kafka-healthcheck-scheduler-") + healthCheckScheduler.startup() + healthCheckScheduler.schedule(name = "halt-broker-if-not-healthy", + fun = haltIfNotHealthy, + period = 10000, + unit = TimeUnit.MILLISECONDS) + // Now that the broker is successfully registered, checkpoint its metadata checkpointBrokerMetadata(ZkMetaProperties(clusterId, config.brokerId)) @@ -660,6 +677,9 @@ class KafkaServer( CoreUtils.swallow(controlledShutdown(), this) _brokerState = BrokerState.SHUTTING_DOWN + if (healthCheckScheduler != null) + healthCheckScheduler.shutdown() + if (dynamicConfigManager != null) CoreUtils.swallow(dynamicConfigManager.shutdown(), this) diff --git a/core/src/main/scala/kafka/server/QuotaFactory.scala b/core/src/main/scala/kafka/server/QuotaFactory.scala index f3901f6b29b11..3fad3d2416c1d 100644 --- a/core/src/main/scala/kafka/server/QuotaFactory.scala +++ b/core/src/main/scala/kafka/server/QuotaFactory.scala @@ -19,6 +19,7 @@ package kafka.server import kafka.server.QuotaType._ import kafka.utils.Logging import org.apache.kafka.common.TopicPartition +import kafka.utils.KafkaScheduler import org.apache.kafka.common.metrics.Metrics import org.apache.kafka.server.quota.ClientQuotaCallback import org.apache.kafka.common.utils.Time @@ -72,13 +73,16 @@ object QuotaFactory extends Logging { } def instantiate(cfg: KafkaConfig, metrics: Metrics, time: Time, threadNamePrefix: String): QuotaManagers = { + instantiate(cfg, metrics, time, None, threadNamePrefix) + } + def instantiate(cfg: KafkaConfig, metrics: Metrics, time: Time, schedulerOpt: Option[KafkaScheduler], threadNamePrefix: String): QuotaManagers = { val clientQuotaCallback = Option(cfg.getConfiguredInstance(KafkaConfig.ClientQuotaCallbackClassProp, classOf[ClientQuotaCallback])) QuotaManagers( - new ClientQuotaManager(clientConfig(cfg), metrics, Fetch, time, threadNamePrefix, clientQuotaCallback), - new ClientQuotaManager(clientConfig(cfg), metrics, Produce, time, threadNamePrefix, clientQuotaCallback), - new ClientRequestQuotaManager(clientConfig(cfg), metrics, time, threadNamePrefix, clientQuotaCallback), + new ClientQuotaManager(clientConfig(cfg), metrics, Fetch, time, schedulerOpt, threadNamePrefix, clientQuotaCallback), + new ClientQuotaManager(clientConfig(cfg), metrics, Produce, time, schedulerOpt, threadNamePrefix, clientQuotaCallback), + new ClientRequestQuotaManager(clientConfig(cfg), metrics, time, schedulerOpt, threadNamePrefix, clientQuotaCallback), new ControllerMutationQuotaManager(clientControllerMutationConfig(cfg), metrics, time, threadNamePrefix, clientQuotaCallback), new ReplicationQuotaManager(replicationConfig(cfg), metrics, LeaderReplication, time), diff --git a/core/src/main/scala/kafka/tools/MirrorMaker.scala b/core/src/main/scala/kafka/tools/MirrorMaker.scala index f6e28656c8dd2..4b55b190a1c6a 100755 --- a/core/src/main/scala/kafka/tools/MirrorMaker.scala +++ b/core/src/main/scala/kafka/tools/MirrorMaker.scala @@ -133,6 +133,7 @@ object MirrorMaker extends Logging with KafkaMetricsGroup { // so if we catch it in commit we can safely retry // and re-throw to break the loop commitOffsets(consumerWrapper) + consumerWrapper.clearOffsetMap() throw e case _: TimeoutException => @@ -151,10 +152,15 @@ object MirrorMaker extends Logging with KafkaMetricsGroup { case _: CommitFailedException => retryNeeded = false + consumerWrapper.clearOffsetMap() warn("Failed to commit offsets because the consumer group has rebalanced and assigned partitions to " + "another instance. If you see this regularly, it could indicate that you need to either increase " + s"the consumer's ${ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG} or reduce the number of records " + s"handled on each iteration with ${ConsumerConfig.MAX_POLL_RECORDS_CONFIG}") + // HOTFIX LIKAFKA-12852 + case e: KafkaException if e.getMessage != null && e.getMessage.contains("may not exist or user may not have Describe access to topic") => + consumerWrapper.clearOffsetMap() + error("Failed to commit offsets due to an unrecoverable error such as committing to a deleted topic", e) } } } else { @@ -348,6 +354,12 @@ object MirrorMaker extends Logging with KafkaMetricsGroup { consumer.commitSync(offsets.map { case (tp, offset) => (tp, new OffsetAndMetadata(offset)) }.asJava) offsets.clear() } + + // HOTFIX LIKAFKA-12852, need to clear the offsets map when a KafkaException has been thrown due to topic + // deletion during offset commet. + def clearOffsetMap(): Unit = { + offsets.clear() + } } private class InternalRebalanceListener(consumerWrapper: ConsumerWrapper, diff --git a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala index 1f7e9d16e758b..603761c8d079e 100644 --- a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala +++ b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala @@ -417,6 +417,36 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness { producer.close() } } + /** + * Test that flush return with TimeoutException when producer is unable to finish sending buffered records in time. +*/ + @Test + def testBoundedFlush(): Unit = { + val producer = createProducer(brokerList) + createTopic(topic, 2, 2) + try { + producer.send(new ProducerRecord(topic, null, "value1".getBytes())).get() + try { + killBroker(0) + } catch { + case _: Throwable => + } + try { + killBroker(1) + } catch { + case _: Throwable => + } + producer.send(new ProducerRecord(topic, null, "value2".getBytes())) + try { + producer.flush(1000, TimeUnit.MILLISECONDS) + fail("TimeoutException should have thrown") + } catch { + case _: TimeoutException => + } + }finally { + producer.close(Duration.ofMillis(1000)) + } + } /** * Test close with zero timeout from caller thread diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala index 4159a1b3c041a..f633df0cd46d7 100644 --- a/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala @@ -17,21 +17,21 @@ package kafka.server import java.net.InetAddress - import kafka.network.RequestChannel.Session import kafka.server.QuotaType._ +import kafka.utils.KafkaScheduler import org.apache.kafka.common.metrics.Quota import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.utils.Sanitizer - import org.junit.jupiter.api.Assertions._ import org.junit.jupiter.api.Test +import org.junit.jupiter.api.{BeforeAll, AfterAll} class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { private val config = ClientQuotaManagerConfig() private def testQuotaParsing(config: ClientQuotaManagerConfig, client1: UserClient, client2: UserClient, randomClient: UserClient, defaultConfigClient: UserClient): Unit = { - val clientQuotaManager = new ClientQuotaManager(config, metrics, Produce, time, "") + val clientQuotaManager = new ClientQuotaManager(config, metrics, Produce, time, Some(ClientQuotaManagerTest.scheduler), "") try { // Case 1: Update the quota. Assert that the new quota value is returned @@ -161,7 +161,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { def testGetMaxValueInQuotaWindowWithNonDefaultQuotaWindow(): Unit = { val numFullQuotaWindows = 3 // 3 seconds window (vs. 10 seconds default) val nonDefaultConfig = ClientQuotaManagerConfig(numQuotaSamples = numFullQuotaWindows + 1) - val clientQuotaManager = new ClientQuotaManager(nonDefaultConfig, metrics, Fetch, time, "") + val clientQuotaManager = new ClientQuotaManager(nonDefaultConfig, metrics, Fetch, time, Some(ClientQuotaManagerTest.scheduler), "") val userSession = Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "userA"), InetAddress.getLocalHost) try { @@ -180,8 +180,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { def testSetAndRemoveDefaultUserQuota(): Unit = { // quotaTypesEnabled will be QuotaTypes.NoQuotas initially val clientQuotaManager = new ClientQuotaManager(ClientQuotaManagerConfig(), - metrics, Produce, time, "") - + metrics, Produce, time, Some(ClientQuotaManagerTest.scheduler), "") try { // no quota set yet, should not throttle checkQuota(clientQuotaManager, "userA", "client1", Long.MaxValue, 1000, false) @@ -202,7 +201,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { def testSetAndRemoveUserQuota(): Unit = { // quotaTypesEnabled will be QuotaTypes.NoQuotas initially val clientQuotaManager = new ClientQuotaManager(ClientQuotaManagerConfig(), - metrics, Produce, time, "") + metrics, Produce, time, Some(ClientQuotaManagerTest.scheduler), "") try { // Set quota config @@ -221,7 +220,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { def testSetAndRemoveUserClientQuota(): Unit = { // quotaTypesEnabled will be QuotaTypes.NoQuotas initially val clientQuotaManager = new ClientQuotaManager(ClientQuotaManagerConfig(), - metrics, Produce, time, "") + metrics, Produce, time, Some(ClientQuotaManagerTest.scheduler), "") try { // Set quota config @@ -239,7 +238,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { @Test def testQuotaConfigPrecedence(): Unit = { val clientQuotaManager = new ClientQuotaManager(ClientQuotaManagerConfig(), - metrics, Produce, time, "") + metrics, Produce, time, Some(ClientQuotaManagerTest.scheduler), "") try { clientQuotaManager.updateQuota(Some(ConfigEntityName.Default), None, None, Some(new Quota(1000, true))) @@ -302,8 +301,9 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { @Test def testQuotaViolation(): Unit = { - val clientQuotaManager = new ClientQuotaManager(config, metrics, Produce, time, "") + val clientQuotaManager = new ClientQuotaManager(config, metrics, Produce, time, Some(ClientQuotaManagerTest.scheduler), "") val queueSizeMetric = metrics.metrics().get(metrics.metricName("queue-size", "Produce", "")) + val throttleCountMetric = metrics.metrics().get(metrics.metricName("throttle-count", "Produce", "")) try { clientQuotaManager.updateQuota(None, Some(ConfigEntityName.Default), Some(ConfigEntityName.Default), Some(new Quota(500, true))) @@ -326,6 +326,8 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { assertEquals(2100, throttleTime, "Should be throttled") throttle(clientQuotaManager, "ANONYMOUS", "unknown", throttleTime, callback) assertEquals(1, queueSizeMetric.metricValue.asInstanceOf[Double].toInt) + assertEquals(1, throttleCountMetric.metricValue.asInstanceOf[Double].toInt) + // After a request is delayed, the callback cannot be triggered immediately clientQuotaManager.throttledChannelReaper.doWork() assertEquals(0, numCallbacks) @@ -333,6 +335,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { // Callback can only be triggered after the delay time passes clientQuotaManager.throttledChannelReaper.doWork() + assertEquals(1, throttleCountMetric.metricValue.asInstanceOf[Double].toInt) assertEquals(0, queueSizeMetric.metricValue.asInstanceOf[Double].toInt) assertEquals(1, numCallbacks) @@ -351,7 +354,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { @Test def testExpireThrottleTimeSensor(): Unit = { - val clientQuotaManager = new ClientQuotaManager(config, metrics, Produce, time, "") + val clientQuotaManager = new ClientQuotaManager(config, metrics, Produce, time, Some(ClientQuotaManagerTest.scheduler), "") try { clientQuotaManager.updateQuota(None, Some(ConfigEntityName.Default), Some(ConfigEntityName.Default), Some(new Quota(500, true))) @@ -373,7 +376,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { @Test def testExpireQuotaSensors(): Unit = { - val clientQuotaManager = new ClientQuotaManager(config, metrics, Produce, time, "") + val clientQuotaManager = new ClientQuotaManager(config, metrics, Produce, time, Some(ClientQuotaManagerTest.scheduler), "") try { clientQuotaManager.updateQuota(None, Some(ConfigEntityName.Default), Some(ConfigEntityName.Default), Some(new Quota(500, true))) @@ -399,7 +402,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { @Test def testClientIdNotSanitized(): Unit = { - val clientQuotaManager = new ClientQuotaManager(config, metrics, Produce, time, "") + val clientQuotaManager = new ClientQuotaManager(config, metrics, Produce, time, Some(ClientQuotaManagerTest.scheduler), "") val clientId = "client@#$%" try { clientQuotaManager.updateQuota(None, Some(ConfigEntityName.Default), Some(ConfigEntityName.Default), @@ -425,3 +428,17 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest { def sanitizedConfigClientId = configClientId.map(x => if (x == ConfigEntityName.Default) ConfigEntityName.Default else Sanitizer.sanitize(x)) } } + +object ClientQuotaManagerTest { + val scheduler = new KafkaScheduler(1) + + @BeforeAll + def startScheduler(): Unit = { + scheduler.startup() + } + + @AfterAll + def shutdownScheduler(): Unit = { + scheduler.shutdown() + } +} diff --git a/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala index db2dceae193d2..27bdf8800120a 100644 --- a/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala @@ -17,17 +17,17 @@ package kafka.server import kafka.server.QuotaType.Request +import kafka.utils.KafkaScheduler import org.apache.kafka.common.metrics.Quota - import org.junit.jupiter.api.Assertions._ import org.junit.jupiter.api.Test class ClientRequestQuotaManagerTest extends BaseClientQuotaManagerTest { private val config = ClientQuotaManagerConfig() - + private val scheduler = new KafkaScheduler(1) @Test def testRequestPercentageQuotaViolation(): Unit = { - val clientRequestQuotaManager = new ClientRequestQuotaManager(config, metrics, time, "", None) + val clientRequestQuotaManager = new ClientRequestQuotaManager(config, metrics, time, Some(scheduler), "", None) clientRequestQuotaManager.updateQuota(Some("ANONYMOUS"), Some("test-client"), Some("test-client"), Some(Quota.upperBound(1))) val queueSizeMetric = metrics.metrics().get(metrics.metricName("queue-size", Request.toString, "")) def millisToPercent(millis: Double) = millis * 1000 * 1000 * ClientRequestQuotaManager.NanosToPercentagePerSecond diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala index 2e38df00f11af..492070478d250 100755 --- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala @@ -615,6 +615,8 @@ class KafkaConfigTest { case KafkaConfig.NumReplicaAlterLogDirsThreadsProp => assertPropertyInvalid(baseProperties, name, "not_a_number") case KafkaConfig.QueuedMaxBytesProp => assertPropertyInvalid(baseProperties, name, "not_a_number") case KafkaConfig.RequestTimeoutMsProp => assertPropertyInvalid(baseProperties, name, "not_a_number") + case KafkaConfig.HeapDumpFolderProp => //ignore string + case KafkaConfig.HeapDumpTimeoutProp => assertPropertyInvalid(baseProperties, name, "not_a_number") case KafkaConfig.ConnectionSetupTimeoutMsProp => assertPropertyInvalid(baseProperties, name, "not_a_number") case KafkaConfig.ConnectionSetupTimeoutMaxMsProp => assertPropertyInvalid(baseProperties, name, "not_a_number") @@ -709,6 +711,9 @@ class KafkaConfigTest { case KafkaConfig.OffsetsRetentionCheckIntervalMsProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "0") case KafkaConfig.OffsetCommitTimeoutMsProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "0") case KafkaConfig.OffsetCommitRequiredAcksProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "-2") + case KafkaConfig.OffsetsTopicMaxMessageBytesProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "-1") + case KafkaConfig.OffsetsTopicMinInSyncReplicasProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "0") + case KafkaConfig.OffsetsTopicMinCompactionLagMsProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "-1") case KafkaConfig.TransactionalIdExpirationMsProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "0", "-2") case KafkaConfig.TransactionsMaxTimeoutMsProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "0", "-2") case KafkaConfig.TransactionsTopicMinISRProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "0", "-2") diff --git a/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala b/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala index 15ad22d97258f..b6c8fc6230d6d 100644 --- a/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala +++ b/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala @@ -51,7 +51,7 @@ class ThrottledChannelExpirationTest { @Test def testCallbackInvocationAfterExpiration(): Unit = { - val clientMetrics = new ClientQuotaManager(ClientQuotaManagerConfig(), metrics, QuotaType.Produce, time, "") + val clientMetrics = new ClientQuotaManager(ClientQuotaManagerConfig(), metrics, QuotaType.Produce, time, None, "") val delayQueue = new DelayQueue[ThrottledChannel]() val reaper = new clientMetrics.ThrottledChannelReaper(delayQueue, "") diff --git a/gradle.properties b/gradle.properties index 9eeca76f3dc9a..0d8fe5897ee0d 100644 --- a/gradle.properties +++ b/gradle.properties @@ -13,15 +13,20 @@ # See the License for the specific language governing permissions and # limitations under the License. -group=org.apache.kafka -# NOTE: When you change this version number, you should also make sure to update -# the version numbers in +group=com.linkedin.kafka + +# NOTE: publishing artifacts requires a property named "version" to be set explicitly. For example, +# ./gradlew -Pversion= publish +# +# You should also make sure to update the version numbers in # - docs/js/templateData.js # - tests/kafkatest/__init__.py # - tests/kafkatest/version.py (variable DEV_VERSION) # - kafka-merge-pr.py version=3.0.1-SNAPSHOT -scalaVersion=2.13.6 +scalaVersion=2.12 task=build org.gradle.jvmargs=-Xmx2g -Xss4m -XX:+UseParallelGC org.gradle.parallel=true +skipSigning=true +jfrogRepoUrl=https://linkedin.jfrog.io/artifactory/kafka