github-actions[bot] commented on code in PR #4381:
URL: https://github.com/apache/iggy/pull/4381#discussion_r4165481148
##########
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.
Review Comment:
nit: The new Javadoc does not say that the active segment survives or that a
committed consumer offset can stop the removal. Add those facts and the
`delete_segments` permission requirement, as `IIggySegment` in the C# SDK does.
##########
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,
Review Comment:
nit: `CompressionAlgorithm` is spelled out fully qualified while the file
imports its other topic types. Import it and use `CompressionAlgorithm.None`.
##########
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));
+ for (int count = 0; count < 5; count++) {
+ client.messages().sendMessages(STREAM_NAME, topicId,
Partitioning.partitionId(0L), List.of(message));
+ }
+
+ var topic = topicsClient.getTopic(STREAM_NAME, topicId).orElseThrow();
+ assertThat(topic.partitions().get(0).segmentsCount()).isEqualTo(2L);
+
+ // when
+ partitionsClient.deleteSegments(STREAM_NAME, topicId, 0L, 1L);
+
+ // then
+ topic = topicsClient.getTopic(STREAM_NAME, topicId).orElseThrow();
+ assertThat(topic.partitions().get(0).segmentsCount()).isEqualTo(1L);
Review Comment:
warning: The test asserts the segment count immediately after
`deleteSegments` returns, but the server acks the metadata commit before the
reconciler removes the segment locally. Poll `getTopic` until the count drops,
the way `await_segment_layout` does in
`core/integration/tests/server/scenarios/purge_delete_scenario.rs:171`.
##########
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());
Review Comment:
warning: `segmentsCount.intValue()` truncates the `Long`, so a negative
count encodes as 4294967295 and the server resolves it to every sealed segment
the consumer barrier allows. Reject counts outside 0..4294967295 before the
write, because the HTTP path already rejects a negative value.
##########
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
+ */
+ CompletableFuture<Void> deleteSegments(StreamId streamId, TopicId topicId,
Long partitionId, Long segmentsCount);
Review Comment:
nit: The new method has no numeric-id overload, but `createPartitions` and
`deletePartitions` do. Add the `(Long, Long, Long, Long)` default to both
interfaces, also at
`foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/PartitionsClient.java:39`.
##########
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:
simplification: The new method inlines `send(...).thenAccept(response ->
response.release())`, and `AsyncTcpConnection.sendAndRelease()` at line 394
does the same work. Call it instead, as `UsersTcpClient` does, and save about 3
lines.
--
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]
