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 11eabc05f6d KAFKA-13716 Add unit tests for DeleteRecordsCommand
(#23018)
11eabc05f6d is described below
commit 11eabc05f6dd18598f8154b8675e21d2ccecd218
Author: Ken Huang <[email protected]>
AuthorDate: Sun Aug 2 02:48:38 2026 +0800
KAFKA-13716 Add unit tests for DeleteRecordsCommand (#23018)
Adds unit tests for DeleteRecordsCommand.
Reviewers: Chia-Ping Tsai <[email protected]>
Co-authored-by: @Dpk376
---
.../kafka/tools/DeleteRecordsCommandTest.java | 86 ++++++++++++++++++++++
1 file changed, 86 insertions(+)
diff --git
a/tools/src/test/java/org/apache/kafka/tools/DeleteRecordsCommandTest.java
b/tools/src/test/java/org/apache/kafka/tools/DeleteRecordsCommandTest.java
index a79f8ef3df3..73e9a8fb1e2 100644
--- a/tools/src/test/java/org/apache/kafka/tools/DeleteRecordsCommandTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/DeleteRecordsCommandTest.java
@@ -18,11 +18,15 @@ package org.apache.kafka.tools;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.DeleteRecordsResult;
+import org.apache.kafka.clients.admin.DeletedRecords;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.test.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
@@ -31,7 +35,9 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.PrintStream;
import java.nio.file.NoSuchFileException;
import java.util.HashMap;
import java.util.List;
@@ -42,6 +48,9 @@ import java.util.Set;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
public class DeleteRecordsCommandTest {
@@ -162,6 +171,83 @@ public class DeleteRecordsCommandTest {
assertEquals(List.of(1L), res.get(new TopicPartition("t", 1)));
}
+ @Test
+ public void testExecuteSuccessOutputWithMockAdmin() throws Exception {
+ TopicPartition tp = new TopicPartition("t", 0);
+
+ KafkaFuture<DeletedRecords> future = KafkaFuture.completedFuture(new
DeletedRecords(5L));
+
+ DeleteRecordsResult deleteResult = mock(DeleteRecordsResult.class);
+ when(deleteResult.lowWatermarks()).thenReturn(Map.of(tp, future));
+
+ Admin mockAdmin = mock(Admin.class);
+ when(mockAdmin.deleteRecords(any())).thenReturn(deleteResult);
+
+ ByteArrayOutputStream bout = new ByteArrayOutputStream();
+ DeleteRecordsCommand.execute(
+ mockAdmin,
+
"{\"partitions\":[{\"topic\":\"t\",\"partition\":0,\"offset\":5}]}",
+ new PrintStream(bout)
+ );
+
+ String output = bout.toString();
+ assertTrue(output.contains("Executing records delete operation"));
+ assertTrue(output.contains("Records delete operation completed:"));
+ assertTrue(output.contains("partition: t-0\tlow_watermark: 5"));
+ }
+
+ @Test
+ public void testExecutePartitionErrorWithMockAdmin() throws Exception {
+ TopicPartition tp = new TopicPartition("t", 0);
+
+ KafkaFutureImpl<DeletedRecords> future = new KafkaFutureImpl<>();
+ future.completeExceptionally(new RuntimeException("Partition not
found"));
+
+ DeleteRecordsResult deleteResult = mock(DeleteRecordsResult.class);
+ when(deleteResult.lowWatermarks()).thenReturn(Map.of(tp, future));
+
+ Admin mockAdmin = mock(Admin.class);
+ when(mockAdmin.deleteRecords(any())).thenReturn(deleteResult);
+
+ ByteArrayOutputStream bout = new ByteArrayOutputStream();
+ DeleteRecordsCommand.execute(
+ mockAdmin,
+
"{\"partitions\":[{\"topic\":\"t\",\"partition\":0,\"offset\":1}]}",
+ new PrintStream(bout)
+ );
+
+ String output = bout.toString();
+ assertTrue(output.contains("partition: t-0\terror:"));
+ assertTrue(output.contains("Partition not found"));
+ }
+
+ @Test
+ public void testExecuteDuplicatePartitionsWithoutCluster() {
+ Admin mockAdmin = mock(Admin.class);
+
+ AdminCommandFailedException exception = assertThrows(
+ AdminCommandFailedException.class,
+ () -> DeleteRecordsCommand.execute(
+ mockAdmin,
+ "{\"partitions\":[" +
+ "{\"topic\":\"t\",\"partition\":0,\"offset\":1}," +
+ "{\"topic\":\"t\",\"partition\":0,\"offset\":2}]}",
+ System.out
+ )
+ );
+
+ assertEquals(
+ "Offset json file contains duplicate topic partitions: t-0",
+ exception.getMessage()
+ );
+ }
+
+ @Test
+ public void testParseInvalidJsonThrows() {
+ assertCommandThrows(AdminOperationException.class, "not-valid-json");
+ assertCommandThrows(AdminOperationException.class, "");
+ }
+
/**
* Asserts that {@link
DeleteRecordsCommand#parseOffsetJsonStringWithoutDedup(String)} throws {@link
AdminOperationException}.
* @param jsonData Data to check.
