mmodzelewski commented on code in PR #4351:
URL: https://github.com/apache/iggy/pull/4351#discussion_r4165497119
##########
foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/IntegrationTest.java:
##########
@@ -117,4 +124,38 @@ protected void trackUser(Long userId) {
createdUserIds.add(userId);
}
}
+
+ protected static PolledMessages pollMessages(
Review Comment:
Warning: `spotlessJavaCheck` fails on this signature, so the Java lint job
fails. Run `./gradlew spotlessApply` and commit the result.
Also at line 145 and `TopicsClientBaseTest.java:81`.
##########
foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/StreamClientBaseTest.java:
##########
@@ -87,4 +95,31 @@ void shouldReturnEmptyForNonExistingStream() {
// then
assertThat(stream).isEmpty();
}
+
+ @Test
+ void shouldPurgeStream() {
Review Comment:
Nit: This test covers one topic and does not assert that the topic still
exists after the purge. Add a second topic with messages and assert that both
remain and are empty.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/StreamsClient.java:
##########
@@ -149,4 +149,29 @@ default CompletableFuture<Void> deleteStream(Long
streamId) {
* @throws org.apache.iggy.exception.IggyException if the stream does not
exist
*/
CompletableFuture<Void> deleteStream(StreamId streamId);
+
+ /**
+ * Purges a stream, deleting all messages in all of its topics and
partitions, by numeric ID.
+ *
+ * <p>This is a convenience overload that wraps the numeric ID into a
{@link StreamId}.
+ *
+ * <p>Unlike {@link #deleteStream(StreamId)}, the stream itself and its
topics are kept.
+ *
+ * @param streamId the numeric stream identifier
+ * @return a {@link CompletableFuture} that completes when the server
accepts the purge
+ */
+ default CompletableFuture<Void> purgeStream(Long streamId) {
+ return purgeStream(StreamId.of(streamId));
+ }
+
+ /**
+ * Purges a stream, deleting all messages in all of its topics and
partitions.
Review Comment:
Nit: The Javadoc does not say that a purge restarts offsets at 0, clears
stored consumer and consumer-group offsets, and cannot be undone. Add these
effects.
Also at `TopicsClient.java:207`.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/TopicsClient.java:
##########
@@ -186,4 +186,32 @@ CompletableFuture<Void> updateTopic(
* @throws org.apache.iggy.exception.IggyException if the topic does not
exist
*/
CompletableFuture<Void> deleteTopic(StreamId streamId, TopicId topicId);
+
+ /**
+ * Purges a topic, deleting all messages in all of its partitions, by
numeric IDs.
+ *
+ * <p>This is a convenience overload that wraps the numeric IDs into a
{@link StreamId} and
+ * a {@link TopicId}.
+ *
+ * <p>Unlike {@link #deleteTopic(StreamId, TopicId)}, the topic itself is
kept.
+ *
+ * @param streamId the numeric stream identifier
+ * @param topicId the numeric topic identifier
+ * @return a {@link CompletableFuture} that completes when the server
accepts the purge
+ */
+ default CompletableFuture<Void> purgeTopic(Long streamId, Long topicId) {
+ return purgeTopic(StreamId.of(streamId), TopicId.of(topicId));
+ }
+
+ /**
+ * Purges a topic, deleting all messages in all of its partitions.
+ *
+ * <p>Unlike {@link #deleteTopic(StreamId, TopicId)}, the topic itself is
kept.
+ *
+ * @param streamId the stream identifier containing the topic
+ * @param topicId the topic identifier to purge
+ * @return a {@link CompletableFuture} that completes when the server
accepts the purge
Review Comment:
Warning: The server wipes the partitions after it acknowledges the purge, so
old messages stay readable briefly and a send made right after the future
completes can be lost. Document this window.
Also at line 200 and `StreamsClient.java:161`, 173.
##########
foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/TopicsClientBaseTest.java:
##########
@@ -64,6 +67,29 @@ void shouldCreateAndDeleteTopic() {
assertThat(topicsClient.getTopics(STREAM_NAME)).isEmpty();
}
+ @Test
+ void shouldPurgeTopic() {
+ // given
+ var topicDetails = topicsClient.createTopic(
+ STREAM_NAME, 1L, CompressionAlgorithm.None, BigInteger.ZERO,
BigInteger.ZERO, "test-topic");
+ var topicId = TopicId.of(topicDetails.id());
+ var messagesClient = client.messages();
+ messagesClient.sendMessages(
+ STREAM_NAME, topicId, Partitioning.partitionId(0L),
List.of(Message.of("message to purge")));
+
+ // The send is acknowledged before this runs, so the message must
already be visible.
+ assertThat(pollMessages(messagesClient, STREAM_NAME,
topicId).messages()).hasSize(1);
+
+ // when
+ topicsClient.purgeTopic(STREAM_NAME, topicId);
+
+ // then
+ assertThat(topicsClient.getTopic(STREAM_NAME, topicId)).isPresent();
+ assertThat(pollUntilEmpty(messagesClient, STREAM_NAME, topicId))
Review Comment:
Nit: No test sends a message after the purge. Send one here and assert that
a poll returns it at offset 0.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java:
##########
@@ -111,4 +111,11 @@ public CompletableFuture<Void> deleteStream(StreamId
streamId) {
return connection().send(CommandCode.Stream.DELETE.getValue(),
payload).thenAccept(ReferenceCounted::release);
}
+
+ @Override
+ public CompletableFuture<Void> purgeStream(StreamId streamId) {
+ var payload = toBytes(streamId);
+
+ return connection().send(CommandCode.Stream.PURGE.getValue(),
payload).thenAccept(ReferenceCounted::release);
Review Comment:
Simplification: Use `connection().sendAndRelease(CommandCode.Stream.PURGE,
payload)` instead of `send(...).thenAccept(ReferenceCounted::release)`.
Also at `TopicsTcpClient.java:219`.
--
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]
