[fix][txn] Enable client without system topics permission to use tran…

…sactions

re-do of 26ef2f8 and
3fb1971 with adjustments to 3.x
codebase.
Also see apache#18718

This change is not in the upstream and likely will not be as per https://lists.apache.org/thread/9wtfnzn6h0glyx5w6w69mmgsokl9xq86

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AllowSystemTransactionTopicLookupAuthorizationProvider.java

@@ -0,0 +1,40 @@
+/*
+ * 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.broker.authorization;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.common.naming.SystemTopicNames;
+import org.apache.pulsar.common.naming.TopicName;
+
+/**
+ * Authorization implementation that allows clients to use transactions even
+ * if they don't have lookup permissions for the system tenant.
+ */
+public class AllowSystemTransactionTopicLookupAuthorizationProvider extends PulsarAuthorizationProvider {
+
+    @Override
+    public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
+                                                     AuthenticationDataSource authenticationData) {
+        if (SystemTopicNames.isTransactionCoordinatorAssign(topicName)) {
+            return CompletableFuture.completedFuture(true);
+        }
+        return super.canLookupAsync(topicName, role, authenticationData);
+    }
+}

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/AuthenticatedTransactionProducerConsumerTest.java

@@ -42,6 +42,7 @@
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
+import org.apache.pulsar.broker.authorization.AllowSystemTransactionTopicLookupAuthorizationProvider;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminBuilder;
 import org.apache.pulsar.client.api.AuthenticationFactory;
@@ -122,6 +123,7 @@ public void setup() throws Exception {
         conf.setProperties(properties);
         conf.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName());
         conf.setBrokerClientAuthenticationParameters("token:" + ADMIN_TOKEN);
+        conf.setAuthorizationProvider(AllowSystemTransactionTopicLookupAuthorizationProvider.class.getName());
         setBrokerCount(1);
         internalSetup();
         setUpBase(1, 1, TOPIC, 1);