KAFKA-17871: avoid blocking the herder thread when producer flushing hangs

.github/scripts/junit.py

@@ -361,7 +361,7 @@ def split_report_path(base_path: str, report_path: str) -> Tuple[str, str]:
         failure_messages.append(f"Gradle task had a failure exit code. Failing this script.")
 
     if thread_dump_url:
-        failure_messages.append(f"Thread dump available at {thread_dump_url}. Failing this script.")
+        failure_messages.append(f"Thread dump available at {thread_dump_url} and the script will now fail.")
 
     if junit_report_url:
         report_md = f"Download [JUnit HTML report]({junit_report_url})"

build.gradle

@@ -369,7 +369,6 @@ subprojects {
 
             // Fix for avoiding inclusion of runtime dependencies marked as 'shadow' in MANIFEST Class-Path.
             // https://github.com/GradleUp/shadow/issues/324
-            afterEvaluate {
               pom.withXml { xml ->
                 def dependenciesNode = xml.asNode().get('dependencies') ?: xml.asNode().appendNode('dependencies')
                 project.configurations.shadowed.allDependencies.each {
@@ -380,7 +379,6 @@ subprojects {
                   dependencyNode.appendNode('scope', 'runtime')
                 }
               }
-            }
           }
 
           afterEvaluate {
@@ -2230,6 +2228,7 @@ project(':storage') {
   }
 
   dependencies {
+    implementation project(':metadata')
     implementation project(':storage:storage-api')
     implementation project(':server-common')
     implementation project(':clients')

checkstyle/import-control-storage.xml

@@ -94,6 +94,8 @@
         <allow pkg="com.yammer.metrics.core" />
         <allow pkg="org.apache.kafka.common" />
         <allow pkg="org.apache.kafka.config" />
+        <allow pkg="org.apache.kafka.image" />
+        <allow pkg="org.apache.kafka.metadata" />
         <allow pkg="org.apache.kafka.server"/>
         <allow pkg="org.apache.kafka.storage.internals"/>
         <allow pkg="org.apache.kafka.storage.log.metrics"/>

checkstyle/suppressions.xml

@@ -98,7 +98,7 @@
               files="(AbstractFetch|ClientTelemetryReporter|ConsumerCoordinator|CommitRequestManager|FetchCollector|OffsetFetcherUtils|KafkaProducer|Sender|ConfigDef|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator|TransactionManager|AbstractStickyAssignor|DefaultSslEngineFactory|Authorizer|RecordAccumulator|MemoryRecords|FetchSessionHandler|MockAdminClient).java"/>
 
     <suppress checks="JavaNCSS"
-              files="(AbstractRequest|AbstractResponse|KerberosLogin|WorkerSinkTaskTest|TransactionManagerTest|SenderTest|KafkaAdminClient|ConsumerCoordinatorTest|KafkaAdminClientTest).java"/>
+              files="(AbstractRequest|AbstractResponse|KerberosLogin|WorkerSinkTaskTest|TransactionManagerTest|SenderTest|KafkaAdminClient|ConsumerCoordinatorTest|KafkaAdminClientTest|KafkaProducerTest).java"/>
 
     <suppress checks="NPathComplexity"
               files="(AbstractMembershipManager|ConsumerCoordinator|BufferPool|MetricName|Node|ConfigDef|RecordBatch|SslFactory|SslTransportLayer|MetadataResponse|KerberosLogin|Selector|Sender|Serdes|TokenInformation|Agent|PluginUtils|MiniTrogdorCluster|TasksRequest|KafkaProducer|AbstractStickyAssignor|Authorizer|FetchSessionHandler|RecordAccumulator|Shell|MockConsumer).java"/>

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java

@@ -122,8 +122,7 @@
         @ClusterConfigProperty(key = "share.coordinator.state.topic.num.partitions", value = "3"),
         @ClusterConfigProperty(key = "share.coordinator.state.topic.replication.factor", value = "1"),
         @ClusterConfigProperty(key = "transaction.state.log.min.isr", value = "1"),
-        @ClusterConfigProperty(key = "transaction.state.log.replication.factor", value = "1"),
-        @ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true")
+        @ClusterConfigProperty(key = "transaction.state.log.replication.factor", value = "1")
     },
     types = {Type.KRAFT}
 )
@@ -1859,8 +1858,7 @@ public void testShareAutoOffsetResetByDurationInvalidFormat() throws Exception {
             @ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "3"),
             @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "3"),
             @ClusterConfigProperty(key = "share.coordinator.state.topic.num.partitions", value = "3"),
-            @ClusterConfigProperty(key = "share.coordinator.state.topic.replication.factor", value = "3"),
-            @ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true")
+            @ClusterConfigProperty(key = "share.coordinator.state.topic.replication.factor", value = "3")
         }
     )
     @Timeout(90)
@@ -2011,8 +2009,7 @@ public void testShareConsumerAfterCoordinatorMovement() throws Exception {
             @ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "3"),
             @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "3"),
             @ClusterConfigProperty(key = "share.coordinator.state.topic.num.partitions", value = "3"),
-            @ClusterConfigProperty(key = "share.coordinator.state.topic.replication.factor", value = "3"),
-            @ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true")
+            @ClusterConfigProperty(key = "share.coordinator.state.topic.replication.factor", value = "3")
         }
     )
     @Timeout(150)

clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java

@@ -108,7 +108,7 @@ public OffsetCommitRequest.Builder buildBatchedRequest(
             .setGroupId(groupId.idValue)
             .setTopics(new ArrayList<>(offsetData.values()));
 
-        return new OffsetCommitRequest.Builder(data);
+        return OffsetCommitRequest.Builder.forTopicNames(data);
     }
 
     @Override

clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteShareGroupOffsetsHandler.java

@@ -99,8 +99,7 @@ DeleteShareGroupOffsetsRequest.Builder buildBatchedRequest(int brokerId, Set<Coo
         return new DeleteShareGroupOffsetsRequest.Builder(
             new DeleteShareGroupOffsetsRequestData()
                 .setGroupId(groupId.idValue)
-                .setTopics(topics),
-            true
+                .setTopics(topics)
         );
     }
 

clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeShareGroupsHandler.java

@@ -92,7 +92,7 @@ public ShareGroupDescribeRequest.Builder buildBatchedRequest(int coordinatorId,
         ShareGroupDescribeRequestData data = new ShareGroupDescribeRequestData()
             .setGroupIds(groupIds)
             .setIncludeAuthorizedOperations(includeAuthorizedOperations);
-        return new ShareGroupDescribeRequest.Builder(data, true);
+        return new ShareGroupDescribeRequest.Builder(data);
     }
 
     @Override

clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java

@@ -104,7 +104,7 @@ public DescribeShareGroupOffsetsRequest.Builder buildBatchedRequest(int coordina
         });
         DescribeShareGroupOffsetsRequestData data = new DescribeShareGroupOffsetsRequestData()
             .setGroups(groups);
-        return new DescribeShareGroupOffsetsRequest.Builder(data, true);
+        return new DescribeShareGroupOffsetsRequest.Builder(data);
     }
 
     @Override

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaShareConsumer.java

@@ -51,8 +51,7 @@
 /**
  * A client that consumes records from a Kafka cluster using a share group.
  * <p>
- *     <em>This is an early access feature under development which is introduced by KIP-932.
- *     It is not suitable for production use until it is fully implemented and released.</em>
+ *     <em>This is a preview feature introduced by KIP-932. It is not yet recommended for production use.</em>
  *
  * <h3>Cross-Version Compatibility</h3>
  * This client can communicate with brokers that are a version that supports share groups. You will receive an
@@ -100,7 +99,7 @@
  * of the topic-partitions that match its subscriptions. Records are acquired for delivery to this consumer with a
  * time-limited acquisition lock. While a record is acquired, it is not available for another consumer. By default,
  * the lock duration is 30 seconds, but it can also be controlled using the group {@code group.share.record.lock.duration.ms}
- * configuration parameter. The idea is that the lock is automatically released once the lock duration has elapsed, and
+ * configuration property. The idea is that the lock is automatically released once the lock duration has elapsed, and
  * then the record is available to be given to another consumer. The consumer which holds the lock can deal with it in
  * the following ways:
  * <ul>
@@ -116,8 +115,8 @@
  * {@code group.share.record.lock.partition.limit}. By limiting the duration of the acquisition lock and automatically
  * releasing the locks, the broker ensures delivery progresses even in the presence of consumer failures.
  * <p>
- * The consumer can choose to use implicit or explicit acknowledgement of the records it processes by configuring the
- * consumer {@code share.acknowledgement.mode} property.
+ * The consumer can choose to use implicit or explicit acknowledgement of the records it processes by using the
+ * consumer {@code share.acknowledgement.mode} configuration property.
  * <p>
  * If the application sets the property to "implicit" or does not set it at all, then the consumer is using
  * <em>implicit acknowledgement</em>. In this mode, the application acknowledges delivery by:
@@ -129,7 +128,7 @@
  *     the delivered records as processed successfully and commits the acknowledgements to Kafka.</li>
  *     <li>Calling {@link #close()} which releases any acquired records without acknowledgement.</li>
  * </ul>
- * If the application sets the property to "explicit", then the consumer is using <em>explicit acknowledgment</em>.
+ * If the application sets the property to "explicit", then the consumer is using <em>explicit acknowledgement</em>.
  * The application must acknowledge all records returned from {@link #poll(Duration)} using
  * {@link #acknowledge(ConsumerRecord, AcknowledgeType)} before its next call to {@link #poll(Duration)}.
  * If the application calls {@link #poll(Duration)} without having acknowledged all records, an
@@ -162,6 +161,7 @@
  *     props.setProperty(&quot;group.id&quot;, &quot;test&quot;);
  *     props.setProperty(&quot;key.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
  *     props.setProperty(&quot;value.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
+ *
  *     KafkaShareConsumer&lt;String, String&gt; consumer = new KafkaShareConsumer&lt;&gt;(props);
  *     consumer.subscribe(Arrays.asList(&quot;foo&quot;));
  *     while (true) {
@@ -181,6 +181,7 @@
  *     props.setProperty(&quot;group.id&quot;, &quot;test&quot;);
  *     props.setProperty(&quot;key.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
  *     props.setProperty(&quot;value.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
+ *
  *     KafkaShareConsumer&lt;String, String&gt; consumer = new KafkaShareConsumer&lt;&gt;(props);
  *     consumer.subscribe(Arrays.asList(&quot;foo&quot;));
  *     while (true) {
@@ -203,6 +204,7 @@
  *     props.setProperty(&quot;key.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
  *     props.setProperty(&quot;value.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
  *     props.setProperty(&quot;share.acknowledgement.mode&quot;, &quot;explicit&quot;);
+ *
  *     KafkaShareConsumer&lt;String, String&gt; consumer = new KafkaShareConsumer&lt;&gt;(props);
  *     consumer.subscribe(Arrays.asList(&quot;foo&quot;));
  *     while (true) {
@@ -443,7 +445,7 @@ public void unsubscribe() {
     }
 
     /**
-     * Fetch data for the topics specified using {@link #subscribe(Collection)}. It is an error to not have
+     * Deliver records for the topics specified using {@link #subscribe(Collection)}. It is an error to not have
      * subscribed to any topics before polling for data.
      *
      * <p>
@@ -452,13 +454,14 @@ public void unsubscribe() {
      *
      * @param timeout The maximum time to block (must not be greater than {@link Long#MAX_VALUE} milliseconds)
      *
-     * @return map of topic to records since the last fetch for the subscribed list of topics
+     * @return map of topic to records
      *
      * @throws AuthenticationException if authentication fails. See the exception for more details
      * @throws AuthorizationException if caller lacks Read access to any of the subscribed
      *             topics or to the share group. See the exception for more details
      * @throws IllegalArgumentException if the timeout value is negative
-     * @throws IllegalStateException if the consumer is not subscribed to any topics
+     * @throws IllegalStateException if the consumer is not subscribed to any topics, or it is using
+     *             explicit acknowledgement and has not acknowledged all records previously delivered
      * @throws ArithmeticException if the timeout is greater than {@link Long#MAX_VALUE} milliseconds.
      * @throws InvalidTopicException if the current subscription contains any invalid
      *             topic (per {@link org.apache.kafka.common.internals.Topic#validate(String)})
@@ -475,11 +478,12 @@ public ConsumerRecords<K, V> poll(Duration timeout) {
      * Acknowledge successful delivery of a record returned on the last {@link #poll(Duration)} call.
      * The acknowledgement is committed on the next {@link #commitSync()}, {@link #commitAsync()} or
      * {@link #poll(Duration)} call.
+     * <p>This method can only be used if the consumer is using <b>explicit acknowledgement</b>.
      *
      * @param record The record to acknowledge
      *
-     * @throws IllegalStateException if the record is not waiting to be acknowledged, or the consumer has already
-     *                               used implicit acknowledgement
+     * @throws IllegalStateException if the record is not waiting to be acknowledged, or the consumer is not using
+     *                               explicit acknowledgement
      */
     @Override
     public void acknowledge(ConsumerRecord<K, V> record) {
@@ -489,14 +493,14 @@ public void acknowledge(ConsumerRecord<K, V> record) {
     /**
      * Acknowledge delivery of a record returned on the last {@link #poll(Duration)} call indicating whether
      * it was processed successfully. The acknowledgement is committed on the next {@link #commitSync()},
-     * {@link #commitAsync()} or {@link #poll(Duration)} call. By using this method, the consumer is using
-     * <b>explicit acknowledgement</b>.
+     * {@link #commitAsync()} or {@link #poll(Duration)} call.
+     * <p>This method can only be used if the consumer is using <b>explicit acknowledgement</b>.
      *
      * @param record The record to acknowledge
      * @param type The acknowledgement type which indicates whether it was processed successfully
      *
-     * @throws IllegalStateException if the record is not waiting to be acknowledged, or the consumer has already
-     *                               used implicit acknowledgement
+     * @throws IllegalStateException if the record is not waiting to be acknowledged, or the consumer is not using
+     *                               explicit acknowledgement
      */
     @Override
     public void acknowledge(ConsumerRecord<K, V> record, AcknowledgeType type) {
@@ -585,7 +589,7 @@ public void setAcknowledgementCommitCallback(AcknowledgementCommitCallback callb
      * client to complete the request.
      * <p>
      * Client telemetry is controlled by the {@link ConsumerConfig#ENABLE_METRICS_PUSH_CONFIG}
-     * configuration option.
+     * configuration property.
      *
      * @param timeout The maximum time to wait for consumer client to determine its client instance ID.
      *                The value must be non-negative. Specifying a timeout of zero means do not

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

@@ -186,7 +186,7 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
      */
     private class BackgroundEventProcessor implements EventProcessor<BackgroundEvent> {
 
-        private Optional<StreamsRebalanceListener> streamsGroupRebalanceCallbacks = Optional.empty();
+        private Optional<StreamsRebalanceListener> streamsRebalanceListener = Optional.empty();
         private final Optional<StreamsRebalanceData> streamsRebalanceData;
 
         public BackgroundEventProcessor() {
@@ -202,7 +202,7 @@ private void setStreamsRebalanceListener(final StreamsRebalanceListener streamsR
                 throw new IllegalStateException("Background event processor was not created to be used with Streams " +
                     "rebalance protocol events");
             }
-            this.streamsGroupRebalanceCallbacks = Optional.of(streamsRebalanceListener);
+            this.streamsRebalanceListener = Optional.of(streamsRebalanceListener);
         }
 
         @Override
@@ -277,20 +277,15 @@ private void processStreamsOnAllTasksLostCallbackNeededEvent(final StreamsOnAllT
 
         private StreamsOnTasksRevokedCallbackCompletedEvent invokeOnTasksRevokedCallback(final Set<StreamsRebalanceData.TaskId> activeTasksToRevoke,
                                                                                          final CompletableFuture<Void> future) {
-            final Optional<KafkaException> error;
-            final Optional<Exception> exceptionFromCallback = streamsGroupRebalanceCallbacks().onTasksRevoked(activeTasksToRevoke);
-            if (exceptionFromCallback.isPresent()) {
-                error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "Task revocation callback throws an error"));
-            } else {
-                error = Optional.empty();
-            }
+            final Optional<Exception> exceptionFromCallback = streamsRebalanceListener().onTasksRevoked(activeTasksToRevoke);
+            final Optional<KafkaException> error = exceptionFromCallback.map(e -> ConsumerUtils.maybeWrapAsKafkaException(e, "Task revocation callback throws an error"));
             return new StreamsOnTasksRevokedCallbackCompletedEvent(future, error);
         }
 
         private StreamsOnTasksAssignedCallbackCompletedEvent invokeOnTasksAssignedCallback(final StreamsRebalanceData.Assignment assignment,
                                                                                            final CompletableFuture<Void> future) {
             final Optional<KafkaException> error;
-            final Optional<Exception> exceptionFromCallback = streamsGroupRebalanceCallbacks().onTasksAssigned(assignment);
+            final Optional<Exception> exceptionFromCallback = streamsRebalanceListener().onTasksAssigned(assignment);
             if (exceptionFromCallback.isPresent()) {
                 error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "Task assignment callback throws an error"));
             } else {
@@ -302,7 +297,7 @@ private StreamsOnTasksAssignedCallbackCompletedEvent invokeOnTasksAssignedCallba
 
         private StreamsOnAllTasksLostCallbackCompletedEvent invokeOnAllTasksLostCallback(final CompletableFuture<Void> future) {
             final Optional<KafkaException> error;
-            final Optional<Exception> exceptionFromCallback = streamsGroupRebalanceCallbacks().onAllTasksLost();
+            final Optional<Exception> exceptionFromCallback = streamsRebalanceListener().onAllTasksLost();
             if (exceptionFromCallback.isPresent()) {
                 error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "All tasks lost callback throws an error"));
             } else {
@@ -318,8 +313,8 @@ private StreamsRebalanceData streamsRebalanceData() {
                     "rebalance protocol events"));
         }
 
-        private StreamsRebalanceListener streamsGroupRebalanceCallbacks() {
-            return streamsGroupRebalanceCallbacks.orElseThrow(
+        private StreamsRebalanceListener streamsRebalanceListener() {
+            return streamsRebalanceListener.orElseThrow(
                 () -> new IllegalStateException("Background event processor was not created to be used with Streams " +
                     "rebalance protocol events"));
         }