hubcio commented on code in PR #4381:
URL: https://github.com/apache/iggy/pull/4381#discussion_r4166358838
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PartitionsTcpClient.java:
##########
@@ -75,4 +75,24 @@ public CompletableFuture<Void> deletePartitions(StreamId
streamId, TopicId topic
response.release();
});
}
+
+ @Override
+ public CompletableFuture<Void> deleteSegments(
+ StreamId streamId, TopicId topicId, Long partitionId, Long
segmentsCount) {
+ var payload = BytesSerializer.toBytes(streamId);
+ payload.writeBytes(BytesSerializer.toBytes(topicId));
+ payload.writeIntLE(partitionId.intValue());
Review Comment:
warning: `partitionId` has the same `intValue()` wrap as `segmentsCount`, so
`4294967296L` goes out as partition 0. range-check both and throw
`IggyInvalidArgumentException`, like `HeaderValue.fromUint32` does.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PartitionsTcpClient.java:
##########
@@ -75,4 +75,24 @@ public CompletableFuture<Void> deletePartitions(StreamId
streamId, TopicId topic
response.release();
});
}
+
+ @Override
+ public CompletableFuture<Void> deleteSegments(
+ StreamId streamId, TopicId topicId, Long partitionId, Long
segmentsCount) {
+ var payload = BytesSerializer.toBytes(streamId);
+ payload.writeBytes(BytesSerializer.toBytes(topicId));
+ payload.writeIntLE(partitionId.intValue());
+ payload.writeIntLE(segmentsCount.intValue());
+
+ log.debug(
+ "Deleting {} segments from partition: {} for stream: {},
topic: {}",
+ segmentsCount,
+ partitionId,
+ streamId,
+ topicId);
+
+ return connection().send(CommandCode.Segment.DELETE.getValue(),
payload).thenAccept(response -> {
Review Comment:
warning: if a caller cancels or times out this future, `thenAccept` never
runs and the response buffer leaks. `sendAndRelease` has the same hole, so this
can wait for a follow-up that releases in an internal stage.
##########
foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/PartitionsClientBaseTest.java:
##########
@@ -60,4 +67,35 @@ void shouldCreateAndDeletePartitions() {
topic = topicsClient.getTopic(STREAM_NAME, TOPIC_NAME).get();
assertThat(topic.partitionsCount()).isEqualTo(1L);
}
+
+ @Test
+ void shouldDeleteSegments() {
+ // given
+ var topicId = TopicId.of("segments-topic");
+ topicsClient.createTopic(
+ STREAM_NAME,
+ 1L,
+ org.apache.iggy.topic.CompressionAlgorithm.None,
+ BigInteger.ZERO,
+ BigInteger.ZERO,
+ "segments-topic",
Review Comment:
nit: pass `topicId.getName()` here instead of repeating `"segments-topic"`.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/PartitionsClient.java:
##########
@@ -72,4 +72,15 @@ default CompletableFuture<Void> deletePartitions(Long
streamId, Long topicId, Lo
* @return A CompletableFuture that completes when the operation is done
*/
CompletableFuture<Void> deletePartitions(StreamId streamId, TopicId
topicId, Long partitionsCount);
+
+ /**
+ * Deletes the oldest sealed segments from a partition asynchronously.
+ *
+ * @param streamId The stream identifier
+ * @param topicId The topic identifier
+ * @param partitionId The partition identifier
+ * @param segmentsCount The number of segments to delete
+ * @return A CompletableFuture that completes when the operation is done
Review Comment:
nit: also say that the future completes at commit, before the segment files
are gone.
##########
foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/PartitionsClientBaseTest.java:
##########
@@ -60,4 +67,35 @@ void shouldCreateAndDeletePartitions() {
topic = topicsClient.getTopic(STREAM_NAME, TOPIC_NAME).get();
assertThat(topic.partitionsCount()).isEqualTo(1L);
}
+
+ @Test
+ void shouldDeleteSegments() {
+ // given
+ var topicId = TopicId.of("segments-topic");
+ topicsClient.createTopic(
+ STREAM_NAME,
+ 1L,
+ org.apache.iggy.topic.CompressionAlgorithm.None,
+ BigInteger.ZERO,
+ BigInteger.ZERO,
+ "segments-topic",
+ TopicOptions.builder()
+ .segmentSize(BigInteger.valueOf(1024 * 1024))
+ .messagesRequiredToSave(1)
+ .build());
+ var message = Message.of("a".repeat(220_000));
Review Comment:
nit: `220_000`, `5` and `1024 * 1024` only work together - five messages
seal one segment and four don't. name them and add a one-line comment with the
math, like `purge_delete_scenario.rs` does.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
