[fix][broker] Keep deduplication snapshots recoverable after skipped or failed attempts

pulsar-broker-common/src/main/java/org/apache/pulsar/common/configuration/anonymizer/DefaultAuthenticationRoleLoggingAnonymizer.java

@@ -62,6 +62,7 @@ public DefaultAuthenticationRoleLoggingAnonymizer(String authenticationRoleLoggi
    }
 
    public String anonymize(String role) {
-      return anonymizerType.anonymize(role);
+      // originalPrincipal is null for clients that do not connect through a proxy
+      return role == null ? null : anonymizerType.anonymize(role);
    }
 }

pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/anonymizer/DefaultAuthenticationRoleLoggingAnonymizerTest.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.configuration.anonymizer;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import org.testng.annotations.Test;
+
+public class DefaultAuthenticationRoleLoggingAnonymizerTest {
+
+    @Test
+    public void testNullRoleIsKeptForEveryType() {
+        // originalPrincipal is null for clients that do not connect through a proxy
+        for (DefaultRoleAnonymizerType type : DefaultRoleAnonymizerType.values()) {
+            assertNull(new DefaultAuthenticationRoleLoggingAnonymizer(type.name()).anonymize(null), type.name());
+        }
+    }
+
+    @Test
+    public void testRoleIsAnonymized() {
+        assertEquals(new DefaultAuthenticationRoleLoggingAnonymizer("NONE").anonymize("role"), "role");
+        assertEquals(new DefaultAuthenticationRoleLoggingAnonymizer("REDACTED").anonymize("role"), "[REDACTED]");
+        assertEquals(new DefaultAuthenticationRoleLoggingAnonymizer("SHA256").anonymize("role").substring(0, 8),
+                "SHA-256:");
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java

@@ -188,6 +188,12 @@ public NotAllowedException(String msg) {
         }
     }
 
+    public static class NotAuthorizedException extends BrokerServiceException {
+        public NotAuthorizedException(String msg) {
+            super(msg);
+        }
+    }
+
     public static class SubscriptionInvalidCursorPosition extends BrokerServiceException {
         public SubscriptionInvalidCursorPosition(String msg) {
             super(msg);
@@ -277,6 +283,8 @@ private static ServerError getClientErrorCode(Throwable t, boolean checkCauseIfU
             return ServerError.InvalidTxnStatus;
         } else if (t instanceof NotAllowedException) {
             return ServerError.NotAllowedError;
+        } else if (t instanceof NotAuthorizedException) {
+            return ServerError.AuthorizationError;
         } else if (t instanceof ProducerFencedException) {
             return ServerError.ProducerFenced;
         } else if (t instanceof TransactionConflictException) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -689,8 +689,8 @@ private CompletableFuture<Boolean> isTopicOperationAllowed(TopicName topicName,
         result.thenAccept(isAuthorized -> {
             if (!isAuthorized) {
                 log.warn()
-                        .attr("authRole", authRole)
-                        .attr("originalPrincipal", originalPrincipal)
+                        .attr("authRole", authenticationRoleLoggingAnonymizer.anonymize(authRole))
+                        .attr("originalPrincipal", authenticationRoleLoggingAnonymizer.anonymize(originalPrincipal))
                         .attr("operation", operation)
                         .attr("topic", topicName)
                         .log("Role or OriginalRole is not authorized to perform operation on topic");
@@ -3540,26 +3540,40 @@ protected void handleGetSchema(CommandGetSchema commandGetSchema) {
         }
 
         final String topic = commandGetSchema.getTopic();
-        String schemaName;
+        final TopicName topicName;
+        final String schemaName;
         try {
-            schemaName = TopicName.get(topic).getSchemaName();
+            topicName = TopicName.get(topic);
+            schemaName = topicName.getSchemaName();
         } catch (Throwable t) {
             commandSender.sendGetSchemaErrorResponse(requestId, ServerError.InvalidTopicName, t.getMessage());
             return;
         }
+        final SchemaVersion requestedVersion = schemaVersion;
 
-        schemaService.getSchema(schemaName, schemaVersion).thenAccept(schemaAndMetadata -> {
-            if (schemaAndMetadata == null) {
-                commandSender.sendGetSchemaErrorResponse(requestId, ServerError.TopicNotFound,
-                        String.format("Topic not found or no-schema %s", topic));
-            } else {
-                commandSender.sendGetSchemaResponse(requestId,
-                        SchemaInfoUtil.newSchemaInfo(schemaName, schemaAndMetadata.schema), schemaAndMetadata.version);
-            }
-        }).exceptionally(ex -> {
-            commandSender.sendGetSchemaErrorResponse(requestId, ServerError.UnknownError, ex.getMessage());
-            return null;
-        });
+        // Producers, consumers and readers fetch the schema of a topic they have looked up, so LOOKUP is a
+        // permission every legitimate caller already holds.
+        isTopicOperationAllowed(topicName, TopicOperation.LOOKUP, authenticationData, originalAuthData)
+                .thenCompose(isAuthorized -> {
+                    if (!isAuthorized) {
+                        commandSender.sendGetSchemaErrorResponse(requestId, ServerError.AuthorizationError,
+                                "Client is not authorized to get the schema of " + topic);
+                        return CompletableFuture.completedFuture(null);
+                    }
+                    return schemaService.getSchema(schemaName, requestedVersion).thenAccept(schemaAndMetadata -> {
+                        if (schemaAndMetadata == null) {
+                            commandSender.sendGetSchemaErrorResponse(requestId, ServerError.TopicNotFound,
+                                    String.format("Topic not found or no-schema %s", topic));
+                        } else {
+                            commandSender.sendGetSchemaResponse(requestId,
+                                    SchemaInfoUtil.newSchemaInfo(schemaName, schemaAndMetadata.schema),
+                                    schemaAndMetadata.version);
+                        }
+                    });
+                }).exceptionally(ex -> {
+                    commandSender.sendGetSchemaErrorResponse(requestId, ServerError.UnknownError, ex.getMessage());
+                    return null;
+                });
     }
 
     @Override
@@ -3568,9 +3582,28 @@ protected void handleGetOrCreateSchema(CommandGetOrCreateSchema commandGetOrCrea
         log.debug("Received CommandGetOrCreateSchema call");
         long requestId = commandGetOrCreateSchema.getRequestId();
         final String topicName = commandGetOrCreateSchema.getTopic();
+        final TopicName parsedTopicName;
+        try {
+            parsedTopicName = TopicName.get(topicName);
+        } catch (Throwable t) {
+            commandSender.sendGetOrCreateSchemaErrorResponse(requestId, ServerError.InvalidTopicName,
+                    t.getMessage());
+            return;
+        }
         SchemaData schemaData = getSchema(commandGetOrCreateSchema.getSchema());
         SchemaData schema = schemaData.getType() == SchemaType.NONE ? null : schemaData;
-        service.getTopicIfExists(topicName).thenAccept(topicOpt -> {
+        // Adding a schema version changes what the topic's producers may send, so it takes PRODUCE, as the
+        // REST schema upload does.
+        CompletableFuture<Optional<Topic>> topicFuture =
+                isTopicOperationAllowed(parsedTopicName, TopicOperation.PRODUCE, authenticationData, originalAuthData)
+                        .thenCompose(isAuthorized -> {
+                            if (!isAuthorized) {
+                                return CompletableFuture.failedFuture(new BrokerServiceException.NotAuthorizedException(
+                                        "Client is not authorized to add a schema to " + topicName));
+                            }
+                            return service.getTopicIfExists(topicName);
+                        });
+        topicFuture.thenAccept(topicOpt -> {
             if (topicOpt.isPresent()) {
                 Topic topic = topicOpt.get();
                 boolean isReplicatorProducer = false;
@@ -3821,9 +3854,10 @@ protected void handleAddPartitionToTxn(CommandAddPartitionToTxn command) {
                     if (!isOwner) {
                         return failedFutureTxnNotOwned(txnID);
                     }
-                    return transactionMetadataStoreService
-                            .addProducedPartitionToTxn(txnID, partitionsList);
+                    return checkTxnPartitionsAuthorized(partitionsList);
                 })
+                .thenCompose(__ -> transactionMetadataStoreService
+                        .addProducedPartitionToTxn(txnID, partitionsList))
                 .whenComplete((v, ex) -> {
                     if (ex == null) {
                         log.debug()
@@ -3844,10 +3878,57 @@ protected void handleAddPartitionToTxn(CommandAddPartitionToTxn command) {
                 });
     }
 
+    /**
+     * The transaction coordinator ends a transaction on its registered partitions with its own identity, so a
+     * client may only register partitions it could produce to.
+     */
+    private CompletableFuture<Void> checkTxnPartitionsAuthorized(List<String> partitions) {
+        List<CompletableFuture<Void>> checks = new ArrayList<>(partitions.size());
+        for (String partition : partitions) {
+            checks.add(checkTxnParticipantAuthorized(partition, null, TopicOperation.PRODUCE));
+        }
+        return FutureUtil.waitForAll(checks);
+    }
+
+    /**
+     * The transaction coordinator ends a transaction on its registered subscriptions with its own identity, so a
+     * client may only register subscriptions it could consume from.
+     */
+    private CompletableFuture<Void> checkTxnSubscriptionsAuthorized(
+            List<org.apache.pulsar.common.api.proto.Subscription> subscriptions) {
+        List<CompletableFuture<Void>> checks = new ArrayList<>(subscriptions.size());
+        for (org.apache.pulsar.common.api.proto.Subscription subscription : subscriptions) {
+            checks.add(checkTxnParticipantAuthorized(subscription.getTopic(), subscription.getSubscription(),
+                    TopicOperation.CONSUME));
+        }
+        return FutureUtil.waitForAll(checks);
+    }
+
+    private CompletableFuture<Void> checkTxnParticipantAuthorized(String topic, String subscription,
+                                                                  TopicOperation operation) {
+        if (!service.isAuthorizationEnabled()) {
+            return CompletableFuture.completedFuture(null);
+        }
+        final TopicName topicName;
+        try {
+            topicName = TopicName.get(topic);
+        } catch (IllegalArgumentException e) {
+            return CompletableFuture.failedFuture(new BrokerServiceException.NotAllowedException(
+                    "Invalid topic name " + topic + ": " + e.getMessage()));
+        }
+        CompletableFuture<Boolean> isAuthorized = subscription == null
+                ? isTopicOperationAllowed(topicName, operation, authenticationData, originalAuthData)
+                : isTopicOperationAllowed(topicName, subscription, operation);
+        return isAuthorized.thenCompose(authorized -> authorized
+                ? CompletableFuture.<Void>completedFuture(null)
+                : CompletableFuture.failedFuture(new BrokerServiceException.NotAuthorizedException(
+                        "Client is not authorized to " + operation + " on " + topic)));
+    }
+
     private CompletableFuture<Void> failedFutureTxnNotOwned(TxnID txnID) {
         String msg = String.format(
                 "Client (%s) is neither the owner of the transaction %s nor a super user",
-                getPrincipal(), txnID
+                authenticationRoleLoggingAnonymizer.anonymize(getPrincipal()), txnID
         );
         log.warn().attr("msg", msg).log("");
         return CompletableFuture.failedFuture(new CoordinatorException.TransactionNotFoundException(msg));
@@ -4254,9 +4335,10 @@ protected void handleAddSubscriptionToTxn(CommandAddSubscriptionToTxn command) {
                     if (!isOwner) {
                         return failedFutureTxnNotOwned(txnID);
                     }
-                    return transactionMetadataStoreService.addAckedPartitionToTxn(txnID,
-                            MLTransactionMetadataStore.subscriptionToTxnSubscription(subscriptionsList));
+                    return checkTxnSubscriptionsAuthorized(subscriptionsList);
                 })
+                .thenCompose(__ -> transactionMetadataStoreService.addAckedPartitionToTxn(txnID,
+                        MLTransactionMetadataStore.subscriptionToTxnSubscription(subscriptionsList)))
                 .whenComplete((v, ex) -> {
                     if (ex == null) {
                         log.debug()

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java

@@ -47,6 +47,7 @@
 import org.apache.pulsar.common.api.proto.MessageMetadata;
 import org.apache.pulsar.common.protocol.Commands;
 import org.apache.pulsar.common.protocol.Markers;
+import org.apache.pulsar.common.util.FutureUtil;
 
 /**
  * Class that contains all the logic to control and perform the deduplication on the broker side.
@@ -593,45 +594,46 @@ public void resetHighestSequenceIdPushed() {
     }
 
     private CompletableFuture<Void> takeSnapshot(Position position) {
-        log.debug("Taking snapshot of sequence ids map");
+        final var cursor = managedCursor;
+        if (cursor == null) {
+            // A publish that already passed isEnabled() can race with disabling deduplication.
+            return CompletableFuture.completedFuture(null);
+        }
 
         if (!snapshotTaking.compareAndSet(false, true)) {
-            log.warn()
+            // The entry/time thresholds can expire again before the previous snapshot completes.
+            // Later triggers will retry; overlapping snapshots must not overwrite newer cursor properties.
+            log.debug()
                     .attr("position", position)
-                    .log("There is a pending snapshot when taking snapshot for");
+                    .log("Skipping deduplication snapshot while another snapshot is pending");
             return CompletableFuture.completedFuture(null);
         }
 
-        Map<String, Long> snapshot = new TreeMap<>();
-        highestSequencedPersisted.forEach((producerName, sequenceId) -> {
-            if (snapshot.size() < maxNumberOfProducers) {
-                snapshot.put(producerName, sequenceId);
+        return FutureUtil.supplySafely(() -> {
+            Map<String, Long> snapshot = new TreeMap<>();
+            highestSequencedPersisted.forEach((producerName, sequenceId) -> {
+                if (snapshot.size() < maxNumberOfProducers) {
+                    snapshot.put(producerName, sequenceId);
+                }
+            });
+            return markDelete(cursor, position, snapshot);
+        }).whenComplete((__, error) -> {
+            try {
+                if (error == null) {
+                    lastSnapshotTimestamp = System.currentTimeMillis();
+                    log.debug()
+                            .attr("position", position)
+                            .log("Stored new deduplication snapshot at");
+                } else {
+                    log.warn()
+                            .attr("position", position)
+                            .exception(error)
+                            .log("Failed to store new deduplication snapshot at");
+                }
+            } finally {
+                snapshotTaking.set(false);
             }
         });
-
-        final var cursor = managedCursor;
-        if (cursor == null) {
-            log.warn()
-                    .attr("position", position)
-                    .log("Cursor is null when taking snapshot for");
-            return CompletableFuture.completedFuture(null);
-        }
-        final var future = markDelete(cursor, position, snapshot).thenRun(() -> {
-            log.debug()
-                    .attr("position", position)
-                    .log("Stored new deduplication snapshot at");
-            lastSnapshotTimestamp = System.currentTimeMillis();
-            snapshotTaking.set(false);
-        });
-        future.exceptionally(e -> {
-            log.warn()
-                    .attr("position", position)
-                    .exception(e)
-                    .log("Failed to store new deduplication snapshot at");
-            snapshotTaking.set(false);
-            return null;
-        });
-        return future;
     }
 
     /**