You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2025/04/02 16:06:33 UTC
(kafka) branch trunk updated: KAFKA-19004 Move DelayedDeleteRecords to server-common module (#19226)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 2994e5eff35 KAFKA-19004 Move DelayedDeleteRecords to server-common module (#19226)
2994e5eff35 is described below
commit 2994e5eff35c7568eda324481e9234e49f1d116f
Author: Xuan-Zhang Gong <go...@gmail.com>
AuthorDate: Thu Apr 3 00:06:27 2025 +0800
KAFKA-19004 Move DelayedDeleteRecords to server-common module (#19226)
Reviewers: Mickael Maison <mi...@gmail.com>, Ken Huang
<s7...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
checkstyle/import-control-server-common.xml | 1 +
core/src/main/scala/kafka/cluster/Partition.scala | 2 +-
.../scala/kafka/server/DelayedDeleteRecords.scala | 134 ---------------------
.../main/scala/kafka/server/ReplicaManager.scala | 38 ++++--
.../scala/unit/kafka/cluster/PartitionTest.scala | 2 +-
.../AbstractCoordinatorConcurrencyTest.scala | 4 +-
.../unit/kafka/server/ReplicaManagerTest.scala | 2 +-
.../server/purgatory/DelayedDeleteRecords.java | 110 +++++++++++++++++
.../purgatory/DeleteRecordsPartitionStatus.java | 59 +++++++++
9 files changed, 205 insertions(+), 147 deletions(-)
diff --git a/checkstyle/import-control-server-common.xml b/checkstyle/import-control-server-common.xml
index 7d7ddb70d87..20f6e32fc9a 100644
--- a/checkstyle/import-control-server-common.xml
+++ b/checkstyle/import-control-server-common.xml
@@ -97,6 +97,7 @@
<subpackage name="purgatory">
<allow pkg="org.apache.kafka.server.metrics" />
<allow pkg="org.apache.kafka.server.util" />
+ <allow pkg="com.yammer.metrics.core" />
</subpackage>
<subpackage name="share">
diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala
index dcca806787c..e816d9c45f8 100755
--- a/core/src/main/scala/kafka/cluster/Partition.scala
+++ b/core/src/main/scala/kafka/cluster/Partition.scala
@@ -42,7 +42,7 @@ import org.apache.kafka.metadata.{LeaderAndIsr, LeaderRecoveryState, MetadataCac
import org.apache.kafka.server.common.RequestLocal
import org.apache.kafka.storage.internals.log.{AppendOrigin, AsyncOffsetReader, FetchDataInfo, LeaderHwChange, LogAppendInfo, LogOffsetMetadata, LogOffsetSnapshot, LogOffsetsListener, LogReadInfo, LogStartOffsetIncrementReason, OffsetResultHolder, UnifiedLog, VerificationGuard}
import org.apache.kafka.server.metrics.KafkaMetricsGroup
-import org.apache.kafka.server.purgatory.{DelayedOperationPurgatory, TopicPartitionOperationKey}
+import org.apache.kafka.server.purgatory.{DelayedDeleteRecords, DelayedOperationPurgatory, TopicPartitionOperationKey}
import org.apache.kafka.server.share.fetch.DelayedShareFetchPartitionKey
import org.apache.kafka.server.storage.log.{FetchIsolation, FetchParams, UnexpectedAppendOffsetException}
import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpoints
diff --git a/core/src/main/scala/kafka/server/DelayedDeleteRecords.scala b/core/src/main/scala/kafka/server/DelayedDeleteRecords.scala
deleted file mode 100644
index c2cffb6a077..00000000000
--- a/core/src/main/scala/kafka/server/DelayedDeleteRecords.scala
+++ /dev/null
@@ -1,134 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kafka.server
-
-
-import kafka.utils.Logging
-
-import java.util.concurrent.TimeUnit
-import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.message.DeleteRecordsResponseData
-import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.requests.DeleteRecordsResponse
-import org.apache.kafka.server.metrics.KafkaMetricsGroup
-import org.apache.kafka.server.purgatory.DelayedOperation
-
-import scala.collection._
-
-
-case class DeleteRecordsPartitionStatus(requiredOffset: Long,
- responseStatus: DeleteRecordsResponseData.DeleteRecordsPartitionResult) {
- @volatile var acksPending = false
-
- override def toString: String = "[acksPending: %b, error: %s, lowWatermark: %d, requiredOffset: %d]"
- .format(acksPending, Errors.forCode(responseStatus.errorCode).toString, responseStatus.lowWatermark, requiredOffset)
-}
-
-/**
- * A delayed delete records operation that can be created by the replica manager and watched
- * in the delete records operation purgatory
- */
-class DelayedDeleteRecords(delayMs: Long,
- deleteRecordsStatus: Map[TopicPartition, DeleteRecordsPartitionStatus],
- replicaManager: ReplicaManager,
- responseCallback: Map[TopicPartition, DeleteRecordsResponseData.DeleteRecordsPartitionResult] => Unit)
- extends DelayedOperation(delayMs) with Logging {
-
- // first update the acks pending variable according to the error code
- deleteRecordsStatus.foreachEntry { (topicPartition, status) =>
- if (status.responseStatus.errorCode == Errors.NONE.code) {
- // Timeout error state will be cleared when required acks are received
- status.acksPending = true
- status.responseStatus.setErrorCode(Errors.REQUEST_TIMED_OUT.code)
- } else {
- status.acksPending = false
- }
-
- trace("Initial partition status for %s is %s".format(topicPartition, status))
- }
-
- /**
- * The delayed delete records operation can be completed if every partition specified in the request satisfied one of the following:
- *
- * 1) There was an error while checking if all replicas have caught up to the deleteRecordsOffset: set an error in response
- * 2) The low watermark of the partition has caught up to the deleteRecordsOffset. set the low watermark in response
- *
- */
- override def tryComplete(): Boolean = {
- // check for each partition if it still has pending acks
- deleteRecordsStatus.foreachEntry { (topicPartition, status) =>
- trace(s"Checking delete records satisfaction for $topicPartition, current status $status")
- // skip those partitions that have already been satisfied
- if (status.acksPending) {
- val (lowWatermarkReached, error, lw) = replicaManager.getPartition(topicPartition) match {
- case HostedPartition.Online(partition) =>
- partition.leaderLogIfLocal match {
- case Some(_) =>
- val leaderLW = partition.lowWatermarkIfLeader
- (leaderLW >= status.requiredOffset, Errors.NONE, leaderLW)
- case None =>
- (false, Errors.NOT_LEADER_OR_FOLLOWER, DeleteRecordsResponse.INVALID_LOW_WATERMARK)
- }
-
- case HostedPartition.Offline(_) =>
- (false, Errors.KAFKA_STORAGE_ERROR, DeleteRecordsResponse.INVALID_LOW_WATERMARK)
-
- case HostedPartition.None =>
- (false, Errors.UNKNOWN_TOPIC_OR_PARTITION, DeleteRecordsResponse.INVALID_LOW_WATERMARK)
- }
- if (error != Errors.NONE || lowWatermarkReached) {
- status.acksPending = false
- status.responseStatus.setErrorCode(error.code)
- status.responseStatus.setLowWatermark(lw)
- }
- }
- }
-
- // check if every partition has satisfied at least one of case A or B
- if (!deleteRecordsStatus.values.exists(_.acksPending))
- forceComplete()
- else
- false
- }
-
- override def onExpiration(): Unit = {
- deleteRecordsStatus.foreachEntry { (topicPartition, status) =>
- if (status.acksPending) {
- DelayedDeleteRecordsMetrics.recordExpiration(topicPartition)
- }
- }
- }
-
- /**
- * Upon completion, return the current response status along with the error code per partition
- */
- override def onComplete(): Unit = {
- val responseStatus = deleteRecordsStatus.map { case (k, status) => k -> status.responseStatus }
- responseCallback(responseStatus)
- }
-}
-
-object DelayedDeleteRecordsMetrics {
- private val metricsGroup = new KafkaMetricsGroup(DelayedDeleteRecordsMetrics.getClass)
-
- private val aggregateExpirationMeter = metricsGroup.newMeter("ExpiresPerSec", "requests", TimeUnit.SECONDS)
-
- def recordExpiration(partition: TopicPartition): Unit = {
- aggregateExpirationMeter.mark()
- }
-}
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
index ad6dfa7fed8..6ff88869d93 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -19,13 +19,14 @@ package kafka.server
import com.yammer.metrics.core.Meter
import kafka.cluster.{Partition, PartitionListener}
import kafka.controller.StateChangeLogger
-import kafka.log.remote.RemoteLogManager
import kafka.log.LogManager
+import kafka.log.remote.RemoteLogManager
import kafka.server.HostedPartition.Online
import kafka.server.QuotaFactory.QuotaManagers
import kafka.server.ReplicaManager.{AtMinIsrPartitionCountMetricName, FailedIsrUpdatesPerSecMetricName, IsrExpandsPerSecMetricName, IsrShrinksPerSecMetricName, LeaderCountMetricName, OfflineReplicaCountMetricName, PartitionCountMetricName, PartitionsWithLateTransactionsCountMetricName, ProducerIdCountMetricName, ReassigningPartitionsMetricName, UnderMinIsrPartitionCountMetricName, UnderReplicatedPartitionsMetricName, createLogReadResult, isListOffsetsTimestampUnsupported}
import kafka.server.share.DelayedShareFetch
import kafka.utils._
+import org.apache.kafka.common.{IsolationLevel, Node, TopicIdPartition, TopicPartition, Uuid}
import org.apache.kafka.common.errors._
import org.apache.kafka.common.internals.{Plugin, Topic}
import org.apache.kafka.common.message.DeleteRecordsResponseData.DeleteRecordsPartitionResult
@@ -46,21 +47,19 @@ import org.apache.kafka.common.requests.FetchRequest.PartitionData
import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
import org.apache.kafka.common.requests._
import org.apache.kafka.common.utils.{Exit, Time, Utils}
-import org.apache.kafka.common.{IsolationLevel, Node, TopicIdPartition, TopicPartition, Uuid}
import org.apache.kafka.image.{LocalReplicaChanges, MetadataImage, TopicsDelta}
import org.apache.kafka.metadata.LeaderConstants.NO_LEADER
import org.apache.kafka.metadata.MetadataCache
-import org.apache.kafka.server.{ActionQueue, DelayedActionQueue, ListOffsetsPartitionStatus, common}
import org.apache.kafka.server.common.{DirectoryEventHandler, RequestLocal, StopPartition, TopicOptionalIdPartition}
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.network.BrokerEndPoint
-import org.apache.kafka.server.purgatory.{DelayedOperationPurgatory, TopicPartitionOperationKey}
+import org.apache.kafka.server.purgatory.{DelayedDeleteRecords, DelayedOperationPurgatory, DeleteRecordsPartitionStatus, TopicPartitionOperationKey}
import org.apache.kafka.server.share.fetch.{DelayedShareFetchKey, DelayedShareFetchPartitionKey}
import org.apache.kafka.server.storage.log.{FetchParams, FetchPartitionData}
import org.apache.kafka.server.util.{Scheduler, ShutdownableThread}
+import org.apache.kafka.server.{ActionQueue, DelayedActionQueue, ListOffsetsPartitionStatus, common}
import org.apache.kafka.storage.internals.checkpoint.{LazyOffsetCheckpoints, OffsetCheckpointFile, OffsetCheckpoints}
-import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchDataInfo, LeaderHwChange, LogAppendInfo, LogConfig, LogDirFailureChannel, LogOffsetMetadata, LogReadInfo, OffsetResultHolder, RecordValidationException, RemoteLogReadResult, RemoteStorageFetchInfo, UnifiedLog => UnifiedLog, VerificationGuard}
-
+import org.apache.kafka.storage.internals.log._
import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import java.io.File
@@ -1314,7 +1313,7 @@ class ReplicaManager(val config: KafkaConfig,
val deleteRecordsStatus = localDeleteRecordsResults.map { case (topicPartition, result) =>
topicPartition ->
- DeleteRecordsPartitionStatus(
+ new DeleteRecordsPartitionStatus(
result.requestedOffset, // requested offset
new DeleteRecordsPartitionResult()
.setLowWatermark(result.lowWatermark)
@@ -1323,8 +1322,31 @@ class ReplicaManager(val config: KafkaConfig,
}
if (delayedDeleteRecordsRequired(localDeleteRecordsResults)) {
+ def onAcks(topicPartition: TopicPartition, status: DeleteRecordsPartitionStatus): Unit = {
+ val (lowWatermarkReached, error, lw) = getPartition(topicPartition) match {
+ case HostedPartition.Online(partition) =>
+ partition.leaderLogIfLocal match {
+ case Some(_) =>
+ val leaderLW = partition.lowWatermarkIfLeader
+ (leaderLW >= status.requiredOffset, Errors.NONE, leaderLW)
+ case None =>
+ (false, Errors.NOT_LEADER_OR_FOLLOWER, DeleteRecordsResponse.INVALID_LOW_WATERMARK)
+ }
+
+ case HostedPartition.Offline(_) =>
+ (false, Errors.KAFKA_STORAGE_ERROR, DeleteRecordsResponse.INVALID_LOW_WATERMARK)
+
+ case HostedPartition.None =>
+ (false, Errors.UNKNOWN_TOPIC_OR_PARTITION, DeleteRecordsResponse.INVALID_LOW_WATERMARK)
+ }
+ if (error != Errors.NONE || lowWatermarkReached) {
+ status.setAcksPending(false)
+ status.responseStatus.setErrorCode(error.code)
+ status.responseStatus.setLowWatermark(lw)
+ }
+ }
// create delayed delete records operation
- val delayedDeleteRecords = new DelayedDeleteRecords(timeout, deleteRecordsStatus, this, responseCallback)
+ val delayedDeleteRecords = new DelayedDeleteRecords(timeout, deleteRecordsStatus.asJava, onAcks, response => responseCallback(response.asScala))
// create a list of (topic, partition) pairs to use as keys for this delayed delete records operation
val deleteRecordsRequestKeys = offsetPerPartition.keys.map(new TopicPartitionOperationKey(_)).toList
diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
index f2454ab34a8..dd2cc239dfc 100644
--- a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
+++ b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
@@ -55,7 +55,7 @@ import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
import org.apache.kafka.coordinator.transaction.TransactionLogConfig
import org.apache.kafka.server.common.{ControllerRequestCompletionHandler, NodeToControllerChannelManager, RequestLocal}
import org.apache.kafka.server.metrics.KafkaYammerMetrics
-import org.apache.kafka.server.purgatory.{DelayedOperationPurgatory, TopicPartitionOperationKey}
+import org.apache.kafka.server.purgatory.{DelayedDeleteRecords, DelayedOperationPurgatory, TopicPartitionOperationKey}
import org.apache.kafka.server.share.fetch.DelayedShareFetchPartitionKey
import org.apache.kafka.server.storage.log.{FetchIsolation, FetchParams, UnexpectedAppendOffsetException}
import org.apache.kafka.server.util.{KafkaScheduler, MockTime}
diff --git a/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala b/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala
index 611cc6dd48e..a3f9354fd8b 100644
--- a/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala
@@ -25,7 +25,7 @@ import kafka.coordinator.AbstractCoordinatorConcurrencyTest._
import kafka.cluster.Partition
import kafka.log.LogManager
import kafka.server.QuotaFactory.QuotaManagers
-import kafka.server.{KafkaConfig, _}
+import kafka.server._
import kafka.utils._
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.protocol.Errors
@@ -33,7 +33,7 @@ import org.apache.kafka.common.record.{MemoryRecords, RecordBatch, RecordValidat
import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.kafka.server.common.RequestLocal
-import org.apache.kafka.server.purgatory.{DelayedOperationPurgatory, TopicPartitionOperationKey}
+import org.apache.kafka.server.purgatory.{DelayedDeleteRecords, DelayedOperationPurgatory, TopicPartitionOperationKey}
import org.apache.kafka.server.util.timer.{MockTimer, Timer}
import org.apache.kafka.server.util.{MockScheduler, MockTime, Scheduler}
import org.apache.kafka.storage.internals.log.{AppendOrigin, LogConfig, UnifiedLog, VerificationGuard}
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index 2c65ad960cb..e014f255d40 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -63,7 +63,7 @@ import org.apache.kafka.server.config.{KRaftConfigs, ReplicationConfigs, ServerL
import org.apache.kafka.server.log.remote.storage._
import org.apache.kafka.server.metrics.{KafkaMetricsGroup, KafkaYammerMetrics}
import org.apache.kafka.server.network.BrokerEndPoint
-import org.apache.kafka.server.purgatory.DelayedOperationPurgatory
+import org.apache.kafka.server.purgatory.{DelayedDeleteRecords, DelayedOperationPurgatory}
import org.apache.kafka.server.share.SharePartitionKey
import org.apache.kafka.server.share.fetch.{DelayedShareFetchGroupKey, DelayedShareFetchKey, ShareFetch}
import org.apache.kafka.server.share.metrics.ShareGroupMetrics
diff --git a/server-common/src/main/java/org/apache/kafka/server/purgatory/DelayedDeleteRecords.java b/server-common/src/main/java/org/apache/kafka/server/purgatory/DelayedDeleteRecords.java
new file mode 100644
index 00000000000..d894a5c6f86
--- /dev/null
+++ b/server-common/src/main/java/org/apache/kafka/server/purgatory/DelayedDeleteRecords.java
@@ -0,0 +1,110 @@
+/*
+ * 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.server.purgatory;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.message.DeleteRecordsResponseData.DeleteRecordsPartitionResult;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.server.metrics.KafkaMetricsGroup;
+
+import com.yammer.metrics.core.Meter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+/**
+ * A delayed delete records operation that can be created by the replica manager and watched
+ * in the delete records operation purgatory
+ */
+public class DelayedDeleteRecords extends DelayedOperation {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DelayedDeleteRecords.class);
+
+ // migration from kafka.server.DelayedDeleteRecordsMetrics
+ private static final KafkaMetricsGroup METRICS_GROUP = new KafkaMetricsGroup("kafka.server", "DelayedDeleteRecordsMetrics");
+ private static final Meter AGGREGATE_EXPIRATION_METER = METRICS_GROUP.newMeter("ExpiresPerSec", "requests",
+ TimeUnit.SECONDS);
+
+ private final Map<TopicPartition, DeleteRecordsPartitionStatus> deleteRecordsStatus;
+ private final BiConsumer<TopicPartition, DeleteRecordsPartitionStatus> onAcksPending;
+ private final Consumer<Map<TopicPartition, DeleteRecordsPartitionResult>> responseCallback;
+
+ public DelayedDeleteRecords(long delayMs,
+ Map<TopicPartition, DeleteRecordsPartitionStatus> deleteRecordsStatus,
+ // To maintain compatibility with dependency packages, the logic has been moved to the caller.
+ BiConsumer<TopicPartition, DeleteRecordsPartitionStatus> onAcksPending,
+ Consumer<Map<TopicPartition, DeleteRecordsPartitionResult>> responseCallback) {
+ super(delayMs);
+ this.onAcksPending = onAcksPending;
+ this.deleteRecordsStatus = Collections.unmodifiableMap(deleteRecordsStatus);
+ this.responseCallback = responseCallback;
+ // first update the acks pending variable according to the error code
+ deleteRecordsStatus.forEach((topicPartition, status) -> {
+ if (status.responseStatus().errorCode() == Errors.NONE.code()) {
+ // Timeout error state will be cleared when required acks are received
+ status.setAcksPending(true);
+ status.responseStatus().setErrorCode(Errors.REQUEST_TIMED_OUT.code());
+ } else {
+ status.setAcksPending(false);
+ }
+
+ LOG.trace("Initial partition status for {} is {}", topicPartition, status);
+ });
+ }
+
+ /**
+ * The delayed delete records operation can be completed if every partition specified in the request satisfied one of the following:
+ *
+ * 1) There was an error while checking if all replicas have caught up to the deleteRecordsOffset: set an error in response
+ * 2) The low watermark of the partition has caught up to the deleteRecordsOffset. set the low watermark in response
+ *
+ */
+ @Override
+ public boolean tryComplete() {
+ // check for each partition if it still has pending acks
+ deleteRecordsStatus.forEach((topicPartition, status) -> {
+ LOG.trace("Checking delete records satisfaction for {}, current status {}", topicPartition, status);
+ // skip those partitions that have already been satisfied
+ if (status.acksPending()) {
+ onAcksPending.accept(topicPartition, status);
+ }
+ });
+ // check if every partition has satisfied at least one of case A or B
+ return deleteRecordsStatus.values().stream().noneMatch(DeleteRecordsPartitionStatus::acksPending) && forceComplete();
+ }
+
+ @Override
+ public void onExpiration() {
+ AGGREGATE_EXPIRATION_METER.mark(deleteRecordsStatus.values().stream().filter(DeleteRecordsPartitionStatus::acksPending).count());
+ }
+
+ /**
+ * Upon completion, return the current response status along with the error code per partition
+ */
+ @Override
+ public void onComplete() {
+ responseCallback.accept(deleteRecordsStatus.entrySet().stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().responseStatus())));
+ }
+}
diff --git a/server-common/src/main/java/org/apache/kafka/server/purgatory/DeleteRecordsPartitionStatus.java b/server-common/src/main/java/org/apache/kafka/server/purgatory/DeleteRecordsPartitionStatus.java
new file mode 100644
index 00000000000..e0b5e607404
--- /dev/null
+++ b/server-common/src/main/java/org/apache/kafka/server/purgatory/DeleteRecordsPartitionStatus.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.purgatory;
+
+
+import org.apache.kafka.common.message.DeleteRecordsResponseData.DeleteRecordsPartitionResult;
+import org.apache.kafka.common.protocol.Errors;
+
+public class DeleteRecordsPartitionStatus {
+ private final long requiredOffset;
+ private final DeleteRecordsPartitionResult responseStatus;
+ private volatile boolean acksPending;
+
+ public DeleteRecordsPartitionStatus(long requiredOffset, DeleteRecordsPartitionResult responseStatus) {
+ this.requiredOffset = requiredOffset;
+ this.responseStatus = responseStatus;
+ this.acksPending = false;
+ }
+
+ public boolean acksPending() {
+ return acksPending;
+ }
+
+ public void setAcksPending(boolean acksPending) {
+ this.acksPending = acksPending;
+ }
+
+
+ public DeleteRecordsPartitionResult responseStatus() {
+ return responseStatus;
+ }
+
+ public long requiredOffset() {
+ return requiredOffset;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("[acksPending: %b, error: %s, lowWatermark: %d, requiredOffset: %d]",
+ acksPending, Errors.forCode(responseStatus.errorCode()).toString(), responseStatus.lowWatermark(),
+ requiredOffset);
+ }
+
+
+}