KAFKA-19004:Move DelayedDeleteRecords to server-common module #19226

gongxuanzhang · 2025-03-18T09:24:27Z

move DeleteRecordsPartitionStatus.scala to server-common module.
scala --> java

Reviewers: Mickael Maison [email protected], Ken Huang [email protected], Chia-Ping Tsai [email protected]

scala --> java

m1a2st

Thanks @gongxuanzhang for this PR, left some comments

m1a2st · 2025-03-18T14:02:37Z

server-common/src/main/java/org/apache/kafka/server/common/DeleteRecordsPartitionStatus.java

+                } else {
+                    status.setAcksPending(false);
+                }
+            });


We should remain this log trace("Initial partition status for %s is %s".format(topicPartition, status))

m1a2st · 2025-03-18T14:05:16Z

server-common/src/main/java/org/apache/kafka/server/common/DeleteRecordsPartitionStatus.java

+        private static final Meter AGGREGATE_EXPIRATION_METER = METRICS_GROUP.newMeter("ExpiresPerSec", "requests",
+                TimeUnit.SECONDS);
+
+        public static void recordExpiration(TopicPartition partition) {


This TopicPartition partition doesn't use, I think we can remove it.

m1a2st · 2025-03-18T14:13:25Z

server-common/src/main/java/org/apache/kafka/server/common/DeleteRecordsPartitionStatus.java

+    }
+
+
+    public DeleteRecordsResponseData.DeleteRecordsPartitionResult getResponseStatus() {


It seems that the get prefix is rarely used in Kafka, should change to responseStatus

m1a2st · 2025-03-18T14:13:48Z

server-common/src/main/java/org/apache/kafka/server/common/DeleteRecordsPartitionStatus.java

+        return responseStatus;
+    }
+
+    public long getRequiredOffset() {


ditto It seems that the get prefix is rarely used in Kafka, should change to requiredOffset

gongxuanzhang · 2025-03-19T01:40:48Z

@m1a2st thanks for your review, i have changed the code.
I think this pr is a migration. we can't delete this parameter

 public static void recordExpiration(TopicPartition partition) {
            AGGREGATE_EXPIRATION_METER.mark();
        }

gongxuanzhang · 2025-03-20T11:09:51Z

@chia7712 PTAL

chia7712

@gongxuanzhang thanks for this patch!

chia7712 · 2025-03-20T16:37:16Z

server-common/src/main/java/org/apache/kafka/server/common/DeleteRecordsPartitionStatus.java

+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+public class DeleteRecordsPartitionStatus {


the main class should be DelayedDeleteRecords rather than DeleteRecordsPartitionStatus

chia7712 · 2025-03-20T16:37:58Z

core/src/main/scala/kafka/server/ReplicaManager.scala

      // create delayed delete records operation
-      val delayedDeleteRecords = new DelayedDeleteRecords(timeout, deleteRecordsStatus, this, responseCallback)
+      val delayedDeleteRecords = new DeleteRecordsPartitionStatus.DelayedDeleteRecords(timeout, deleteRecordsStatus.asJava,onAcks , responseCallbackJava)


deleteRecordsStatus.asJava, onAcks, responseCallbackJava

chia7712 · 2025-03-20T16:43:58Z

core/src/main/scala/kafka/server/ReplicaManager.scala

+          status.responseStatus.setLowWatermark(lw)
+        }
+      }
+      val responseCallbackJava: java.util.function.Consumer[util.Map[TopicPartition, DeleteRecordsPartitionResult]] =


It seems we don't need to create this object, right? we can pass response => responseCallback(response.asScala) directly

chia7712 · 2025-03-20T16:47:44Z

server-common/src/main/java/org/apache/kafka/server/common/DeleteRecordsPartitionStatus.java

+        }
+    }
+
+    private static class DelayedDeleteRecordsMetrics {


we don't need this static class as all static variables can be moved to DelayedDeleteRecords

chia7712 · 2025-03-20T16:48:34Z

server-common/src/main/java/org/apache/kafka/server/common/DeleteRecordsPartitionStatus.java

+    }
+
+    private static class DelayedDeleteRecordsMetrics {
+        private static final KafkaMetricsGroup METRICS_GROUP = new KafkaMetricsGroup(DelayedDeleteRecordsMetrics.class);


this break the compatibility of metrics name, so please use "kafka.server.DelayedDeleteRecordsMetrics" instead

chia7712 · 2025-03-20T16:49:13Z

server-common/src/main/java/org/apache/kafka/server/common/DeleteRecordsPartitionStatus.java

+            deleteRecordsStatus.forEach((k, status) -> {
+                responseStatus.put(k, status.responseStatus());
+            });
+            responseCallback.accept(responseStatus);


responseCallback.accept(deleteRecordsStatus.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().responseStatus())));

chia7712 · 2025-03-20T16:50:19Z

server-common/src/main/java/org/apache/kafka/server/common/DeleteRecordsPartitionStatus.java

+    private volatile boolean acksPending;
+
+    public DeleteRecordsPartitionStatus(long requiredOffset,
+                                        DeleteRecordsResponseData.DeleteRecordsPartitionResult responseStatus) {


could you please import DeleteRecordsResponseData.DeleteRecordsPartitionResult?

gongxuanzhang · 2025-03-21T16:33:05Z

@chia7712 PTAL

chia7712 · 2025-03-21T18:22:57Z

server-common/src/main/java/org/apache/kafka/server/common/DelayedDeleteRecords.java

+ */
+public class DelayedDeleteRecords extends DelayedOperation {
+
+    private static final Logger log = LoggerFactory.getLogger(DelayedDeleteRecords.class);


chia7712 · 2025-03-21T18:25:58Z

server-common/src/main/java/org/apache/kafka/server/common/DelayedDeleteRecords.java

+                                Consumer<Map<TopicPartition, DeleteRecordsPartitionResult>> responseCallback) {
+        super(delayMs);
+        this.onAcksPending = onAcksPending;
+        this.deleteRecordsStatus = new ConcurrentHashMap<>(deleteRecordsStatus);


why to copy deleteRecordsStatus? it is immutable, right?

I'll make it immutable and avoid copying. Thanks

gongxuanzhang · 2025-03-23T08:52:29Z

@chia7712 PTAL

m1a2st

Thanks for this patch, left some comments

m1a2st · 2025-03-25T02:58:39Z

core/src/main/scala/kafka/server/ReplicaManager.scala

-import kafka.server.ReplicaManager.{AtMinIsrPartitionCountMetricName, FailedIsrUpdatesPerSecMetricName, IsrExpandsPerSecMetricName, IsrShrinksPerSecMetricName, LeaderCountMetricName, OfflineReplicaCountMetricName, PartitionCountMetricName, PartitionsWithLateTransactionsCountMetricName, ProducerIdCountMetricName, ReassigningPartitionsMetricName, UnderMinIsrPartitionCountMetricName, UnderReplicatedPartitionsMetricName, createLogReadResult, isListOffsetsTimestampUnsupported}
+import kafka.server.ReplicaManager._


I think this change is unnecessary.

Let's keep the explicit imports

m1a2st · 2025-03-25T02:58:52Z

core/src/main/scala/kafka/server/ReplicaManager.scala

-import org.apache.kafka.server.common.{DirectoryEventHandler, RequestLocal, StopPartition, TopicOptionalIdPartition}
+import org.apache.kafka.server.common._


I think this change is unnecessary.

m1a2st · 2025-03-25T02:59:09Z

core/src/main/scala/kafka/server/ReplicaManager.scala

-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._


I think this change is unnecessary.

mimaison

Thanks for the PR. I left a few comments.

mimaison · 2025-03-25T14:53:12Z

server-common/src/main/java/org/apache/kafka/server/common/DelayedDeleteRecords.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.common;


Would the org.apache.kafka.server.purgatory package be a better place for these classes? So it's next to the abstract class DelayedOperation it's extending.

+1 to org.apache.kafka.server.purgatory

mimaison · 2025-03-25T14:54:09Z

server-common/src/main/java/org/apache/kafka/server/common/DelayedDeleteRecords.java

+ */
+public class DelayedDeleteRecords extends DelayedOperation {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DelayedDeleteRecords.class);


Should we keep kafka.server.DelayedDeleteRecords as the logger name?

I don't think changing the name will affect anything. Do we have a special need to keep the name

Logger names are what users set in their log4j2 configuration. But it seems we've not consistently kept names for other classes we moved, so this is probably fine.

thanks for your answer

mimaison · 2025-03-25T14:55:25Z

core/src/main/scala/kafka/server/ReplicaManager.scala

-import kafka.server.ReplicaManager.{AtMinIsrPartitionCountMetricName, FailedIsrUpdatesPerSecMetricName, IsrExpandsPerSecMetricName, IsrShrinksPerSecMetricName, LeaderCountMetricName, OfflineReplicaCountMetricName, PartitionCountMetricName, PartitionsWithLateTransactionsCountMetricName, ProducerIdCountMetricName, ReassigningPartitionsMetricName, UnderMinIsrPartitionCountMetricName, UnderReplicatedPartitionsMetricName, createLogReadResult, isListOffsetsTimestampUnsupported}
+import kafka.server.ReplicaManager._


Let's keep the explicit imports

gongxuanzhang · 2025-03-28T17:51:36Z

I changed this pr @mimaison @chia7712 PTAL

chia7712

@gongxuanzhang thanks for updating patch.

chia7712 · 2025-03-31T08:16:02Z

core/src/main/scala/kafka/server/ReplicaManager.scala

      // 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))


sJava, onAcks , response = -> sJava, onAcks, response =

this is not addressed

chia7712 · 2025-03-31T08:20:09Z

server-common/src/main/java/org/apache/kafka/server/purgatory/DelayedDeleteRecords.java

+
+    @Override
+    public void onExpiration() {
+        deleteRecordsStatus.forEach((topicPartition, status) -> {


AGGREGATE_EXPIRATION_METER.mark(deleteRecordsStatus.values().stream().filter(DeleteRecordsPartitionStatus::acksPending).count());

@gongxuanzhang WDYT?

excellent！

chia7712 · 2025-03-31T08:21:27Z

core/src/main/scala/kafka/server/DelayedDeleteRecords.scala

-      status.acksPending = false
-    }
-
-    trace("Initial partition status for %s is %s".format(topicPartition, status))


please keep this log

mimaison · 2025-03-31T17:24:27Z

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" />


Should this be com.yammer.metrics.core?

gongxuanzhang · 2025-04-02T02:22:01Z

@chia7712 PTAL

…#19226) Reviewers: Mickael Maison <[email protected]>, Ken Huang <[email protected]>, Chia-Ping Tsai <[email protected]>

mimaison · 2025-04-07T07:56:22Z

I was looking at #19390 and wondering whether we should put these classes in server instead of server-common as they don't seem to be used elsewhere. For example we put DelayedRemoteListOffsets in the storage module. WDYT?

chia7712 · 2025-04-07T11:39:42Z

I was looking at #19390 and wondering whether we should put these classes in server instead of server-common as they don't seem to be used elsewhere. For example we put DelayedRemoteListOffsets in the storage module. WDYT?

It seems RemoteLogManager could use those Purgatory-related code. I guess RemoteLogManager will be moved to storage module in the future? If so, those Purgatory code should be in server-common module

mimaison · 2025-04-07T12:46:39Z

RemoteLogManager only uses DelayedRemoteListOffsets hence why we put that in the storage module. But as far as I can tell other purgatory code could be in server.

chia7712 · 2025-04-07T12:52:25Z

RemoteLogManager only uses DelayedRemoteListOffsets hence why we put that in the storage module. But as far as I can tell other purgatory code could be in server.

RemoteLogManager also uses TopicPartitionOperationKey (and DelayedOperationKey), so I guess you meant other classes, such as DeleteRecordsPartitionStatus, DelayedDeleteRecords, and DelayedOperation, can be moved to server module, right?

mimaison · 2025-04-07T13:35:55Z

Yes I meant DelayedDeleteRecords, DeleteRecordsPartitionStatus and future classes not needed in a different module, for example the ones in #19390

We need to keep DelayedOperation in server-common as we have implementations in other modules, for example DelayedRemoteListOffsets in storage.

chia7712 · 2025-04-07T13:37:20Z

@gongxuanzhang WDYT? If you agree @mimaison's comment, could you please create a MINOR to address it?

gongxuanzhang · 2025-04-07T13:45:27Z

@mimaison @chia7712
Thanks for your comments
I will create a MINOR PR to address it

move DeleteRecordsPartitionStatus.scala to server-common module.

5a55640

scala --> java

github-actions bot added triage PRs from the community core Kafka Broker labels Mar 18, 2025

chia7712 added the ci-approved label Mar 18, 2025

m1a2st reviewed Mar 18, 2025

View reviewed changes

remain commit

e4e2341

github-actions bot removed the triage PRs from the community label Mar 19, 2025

chia7712 reviewed Mar 20, 2025

View reviewed changes

gongxuanzhang added 3 commits March 21, 2025 10:52

Merge branch 'apache:trunk' into move_deleteRecord

7cbd93e

refactor

e236f2a

refactor import

87bc32c

chia7712 reviewed Mar 21, 2025

View reviewed changes

gongxuanzhang added 2 commits March 23, 2025 14:29

Merge branch 'apache:trunk' into move_deleteRecord

b572a13

format

5e0a336

m1a2st reviewed Mar 25, 2025

View reviewed changes

mimaison reviewed Mar 25, 2025

View reviewed changes

gongxuanzhang added 2 commits March 28, 2025 09:29

refactor import

d546401

optimize import

4529c33

chia7712 mentioned this pull request Mar 28, 2025

KAFKA-14523: Move DelayedRemoteListOffsets to the storage module #19285

Merged

chia7712 reviewed Mar 31, 2025

View reviewed changes

Merge branch 'apache:trunk' into move_deleteRecord

0714951

mimaison reviewed Mar 31, 2025

View reviewed changes

refactor

ea265cd

chia7712 approved these changes Apr 2, 2025

View reviewed changes

chia7712 merged commit 2994e5e into apache:trunk Apr 2, 2025
23 checks passed

gongxuanzhang mentioned this pull request Apr 7, 2025

MINOR: Move DelayedRecords purgatory classes to server #19408

Merged

		}


		public DeleteRecordsResponseData.DeleteRecordsPartitionResult getResponseStatus() {

		import kafka.server.ReplicaManager.{AtMinIsrPartitionCountMetricName, FailedIsrUpdatesPerSecMetricName, IsrExpandsPerSecMetricName, IsrShrinksPerSecMetricName, LeaderCountMetricName, OfflineReplicaCountMetricName, PartitionCountMetricName, PartitionsWithLateTransactionsCountMetricName, ProducerIdCountMetricName, ReassigningPartitionsMetricName, UnderMinIsrPartitionCountMetricName, UnderReplicatedPartitionsMetricName, createLogReadResult, isListOffsetsTimestampUnsupported}
		import kafka.server.ReplicaManager._

		import org.apache.kafka.server.common.{DirectoryEventHandler, RequestLocal, StopPartition, TopicOptionalIdPartition}
		import org.apache.kafka.server.common._

		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._

KAFKA-19004:Move DelayedDeleteRecords to server-common module #19226

KAFKA-19004:Move DelayedDeleteRecords to server-common module #19226

Conversation

gongxuanzhang commented Mar 18, 2025 • edited by chia7712 Loading

m1a2st left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gongxuanzhang commented Mar 19, 2025

gongxuanzhang commented Mar 20, 2025

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gongxuanzhang commented Mar 21, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gongxuanzhang commented Mar 23, 2025

m1a2st left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mimaison left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mimaison Mar 28, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gongxuanzhang commented Mar 28, 2025 • edited Loading

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gongxuanzhang commented Apr 2, 2025

mimaison commented Apr 7, 2025

chia7712 commented Apr 7, 2025

mimaison commented Apr 7, 2025

chia7712 commented Apr 7, 2025

mimaison commented Apr 7, 2025 • edited Loading

chia7712 commented Apr 7, 2025

gongxuanzhang commented Apr 7, 2025

gongxuanzhang commented Mar 18, 2025 •

edited by chia7712

Loading

mimaison Mar 28, 2025 •

edited

Loading

gongxuanzhang commented Mar 28, 2025 •

edited

Loading

mimaison commented Apr 7, 2025 •

edited

Loading