fix: warnings on topic unsubscribe (#2463)

Signed-off-by: emiliyank <e.kadiyski@gmail.com>

Author: emiliyank

sdk/src/main/java/com/hedera/hashgraph/sdk/TopicMessageQuery.java

@@ -320,7 +320,11 @@ public void onNext(ConsensusTopicResponse consensusTopicResponse) {
 
             @Override
             public void onError(Throwable t) {
-                if (attempt >= maxAttempts || !retryHandler.test(t) || cancelledByClient.get()) {
+                if (cancelledByClient.get()) {
+                    return;
+                }
+
+                if (attempt >= maxAttempts || !retryHandler.test(t)) {
                     errorHandler.accept(t, null);
                     return;
                 }

sdk/src/testIntegration/java/com/hedera/hashgraph/sdk/test/integration/TopicMessageIntegrationTest.java

@@ -8,6 +8,7 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 
@@ -123,4 +124,52 @@ void canReceiveALargeTopicMessage() throws Exception {
                     .getReceipt(testEnv.client);
         }
     }
+
+    @Test
+    @DisplayName("Unsubscribing does not log retry warnings")
+    void unsubscribingDoesNotLogRetryWarnings() throws Exception {
+        try (var testEnv = new IntegrationTestEnv(1)) {
+
+            var response = new TopicCreateTransaction()
+                    .setAdminKey(testEnv.operatorKey)
+                    .setTopicMemo("[e2e::TopicCreateTransaction]")
+                    .execute(testEnv.client);
+
+            var topicId = Objects.requireNonNull(response.getReceipt(testEnv.client).topicId);
+
+            var receivedMessage = new AtomicBoolean(false);
+            var retryWarningLogged = new AtomicBoolean(false);
+            var errorHandlerInvoked = new AtomicBoolean(false);
+
+            var retryHandler = new java.util.function.Predicate<Throwable>() {
+                @Override
+                public boolean test(Throwable throwable) {
+                    retryWarningLogged.set(true);
+                    return false; // Don't actually retry
+                }
+            };
+
+            var handle = new TopicMessageQuery()
+                    .setTopicId(topicId)
+                    .setStartTime(Instant.EPOCH)
+                    .setRetryHandler(retryHandler)
+                    .setErrorHandler((throwable, topicMessage) -> errorHandlerInvoked.set(true))
+                    .subscribe(testEnv.client, (message) -> {
+                        receivedMessage.set(true);
+                    });
+
+            handle.unsubscribe();
+
+            Thread.sleep(3000);
+
+            assertThat(retryWarningLogged.get()).isFalse();
+            assertThat(receivedMessage.get()).isFalse();
+            assertThat(errorHandlerInvoked.get()).isFalse();
+
+            new TopicDeleteTransaction()
+                    .setTopicId(topicId)
+                    .execute(testEnv.client)
+                    .getReceipt(testEnv.client);
+        }
+    }
 }