You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/01 18:11:02 UTC
(pulsar) 12/38: [fix][broker] Spurious ERROR log for 307 redirect in getReplicatedSubscriptionStatus (#26706)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-branch-4.2-backports-2026-10-01
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e8a6333e6354e465946fd8a84de51ea7fa9b55ee
Author: Vyacheslav <25...@gmail.com>
AuthorDate: Tue Sep 29 01:23:18 2026 +0200
[fix][broker] Spurious ERROR log for 307 redirect in getReplicatedSubscriptionStatus (#26706)
Co-authored-by: Vyacheslav SHABLISTYY <vy...@rbinternational.com>
Co-authored-by: Lari Hotari <lh...@users.noreply.github.com>
(cherry picked from commit 776786b6d1c45cd8d5f014dba5a7168c4dbfc030)
---
.../broker/admin/impl/PersistentTopicsBase.java | 6 ++-
.../pulsar/broker/admin/PersistentTopicsTest.java | 54 ++++++++++++++++++++++
2 files changed, 58 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 758357a58feb..8cf6047e9bf3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -5120,8 +5120,10 @@ public class PersistentTopicsBase extends AdminResource {
.whenComplete((res, e) -> {
if (e != null) {
Throwable cause = FutureUtil.unwrapCompletionException(e);
- log.error("[{}] Failed to get replicated subscription status on {} {}", clientAppId(),
- topicName, subName, cause);
+ if (isNot307And404Exception(cause)) {
+ log.error("[{}] Failed to get replicated subscription status on {} {}", clientAppId(),
+ topicName, subName, cause);
+ }
resumeAsyncResponseExceptionally(asyncResponse, e);
} else {
asyncResponse.resume(res);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
index 8d973ac6b285..f647d3fad61c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
@@ -37,6 +37,7 @@ import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
import java.lang.reflect.Field;
+import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
@@ -60,6 +61,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.logging.log4j.Level;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.admin.v2.ExtPersistentTopics;
import org.apache.pulsar.broker.admin.v2.NonPersistentTopics;
@@ -109,6 +111,7 @@ import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.metadata.api.MetadataStoreException;
+import org.apache.pulsar.utils.TestLogAppender;
import org.apache.zookeeper.KeeperException;
import org.awaitility.Awaitility;
import org.mockito.ArgumentCaptor;
@@ -1908,6 +1911,57 @@ public class PersistentTopicsTest extends MockedPulsarServiceBaseTest {
verify(topics, times(1)).getReplicatedSubscriptionStatusAsync(any(), any());
}
+ @Test
+ public void testGetReplicatedSubscriptionStatusDoesNotLogRedirectAsError() throws Exception {
+ // A 307 means this broker does not own the partition and the caller is being sent to the one that
+ // does. That is normal routing rather than a failure, so it must not be reported at ERROR level: a
+ // broker that fans out per-partition admin calls would otherwise log a stack trace per redirect.
+ // The logger is the one inherited from PulsarWebResource, not PersistentTopicsBase: the inherited
+ // field is derived via LOG.with().build(), and slog keeps the parent's name when deriving.
+ @Cleanup
+ TestLogAppender logAppender = TestLogAppender.create(PulsarWebResource.class);
+
+ String topicName = "persistent://" + testTenant + "/" + testNamespaceLocal
+ + "/testGetReplicatedSubscriptionStatusDoesNotLogRedirectAsError";
+ String subName = "sub_testGetReplicatedSubscriptionStatusDoesNotLogRedirectAsError";
+ TopicName partition = TopicName.get(topicName).getPartition(0);
+ admin.topics().createPartitionedTopic(topicName, 1);
+ admin.topics().createSubscription(topicName, subName, MessageId.latest);
+
+ // validateTopicOwnershipAsync() builds the redirect location from the request URI.
+ uriField.set(persistentTopics, uriInfo);
+ doReturn(URI.create(pulsar.getWebServiceAddress() + "/admin/v2/persistent/" + testTenant + "/"
+ + testNamespaceLocal + "/" + partition.getLocalName() + "/subscription/" + subName
+ + "/replicatedSubscriptionStatus")).when(uriInfo).getRequestUri();
+
+ // Pretend the partition moved to another broker between the lookup and this call.
+ NamespaceService namespaceService = pulsar.getNamespaceService();
+ doReturn(CompletableFuture.completedFuture(false))
+ .when(namespaceService).isServiceUnitOwnedAsync(partition);
+ doReturn(namespaceService).when(pulsar).getNamespaceService();
+
+ logAppender.clearEvents();
+
+ // Passing the partition name directly routes straight into
+ // internalGetReplicatedSubscriptionStatusForNonPartitionedTopic, whose handler is the one under
+ // test. The outer handler in internalGetReplicatedSubscriptionStatus is not chained to this call,
+ // so an ERROR captured here can only have come from the inner one.
+ AsyncResponse response = mock(AsyncResponse.class);
+ persistentTopics.getReplicatedSubscriptionStatus(response, testTenant, testNamespaceLocal,
+ partition.getLocalName(), subName, false);
+
+ ArgumentCaptor<WebApplicationException> captor = ArgumentCaptor.forClass(WebApplicationException.class);
+ verify(response, timeout(5000).times(1)).resume(captor.capture());
+ assertEquals(captor.getValue().getResponse().getStatus(),
+ Response.Status.TEMPORARY_REDIRECT.getStatusCode());
+
+ assertTrue(logAppender.getEvents().stream()
+ .noneMatch(event -> event.getLevel() == Level.ERROR
+ && event.getMessage().toString()
+ .contains("Failed to get replicated subscription status")),
+ "A 307 redirect must not be logged at ERROR level");
+ }
+
@Test
public void testNamespaceResources() throws Exception {
String ns1V1 = testNamespace + "v1";