Skip to content

KAFKA-13716: Add unit tests for DeleteRecordsCommand #22573

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Dpk376 wants to merge 3 commits into
base: trunk
Choose a base branch
from
+86 −0
Open

KAFKA-13716: Add unit tests for DeleteRecordsCommand #22573

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ea45028
KAFKA-13716: Add unit tests for DeleteRecordsCommand
Dpk376 Jun 15, 2026
0ab25f5
KAFKA-13716: Fix Spotless formatting
Dpk376 Jun 15, 2026
5499351
KAFKA-13716: assert error messages in DeleteRecordsCommand tests
Dpk376 Jun 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 86 additions & 0 deletions tools/src/test/java/org/apache/kafka/tools/DeleteRecordsCommandTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,11 +18,15 @@

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;
Expand All @@ -31,7 +35,9 @@

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;
Expand All @@ -42,6 +48,9 @@
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 {

Expand Down Expand Up @@ -162,6 +171,83 @@ public void testParse() throws Exception {
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,

@chia7712 chia7712 Jun 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would you mind checking the error message as well?

@Dpk376 Dpk376 Jun 22, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch — the duplicate-partition test now asserts the full AdminCommandFailedException message, and I also tightened the per-partition error test to check the printed error text (Partition not found). Thanks for the review, @chia7712!

() -> 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.
Expand Down