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/20 21:41:14 UTC
(pulsar) branch branch-5.0 updated (ed3507bb5073 -> 48f850be5f1f)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from ed3507bb5073 [fix][ml] Fix race in ManagedCursorImpl skip and getNthEntry that could skip unacknowledged entries (#20488)
new 0ce235304cd9 [fix][broker] Fix the broker's WebSocket service ignoring configurationStoreConfigPath (#27015)
new ee9a0acd6f50 [improve][broker] Enable Log4j2 async loggers again (#26979)
new bb0369c5224c [fix][test] Fix flaky PulsarLedgerUnderreplicationManagerCloseTest.testCloseAggregatesFailuresAndRetriesOnlyUnreleasedLocks (#27013)
new 4b8e30d188ec [improve][doc] Note in CODING.md that a delayedExecutor() task cannot be taken back (#26988)
new 48f850be5f1f [improve][broker] Let authorization providers restrict topic auto-creation (#17411)
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CODING.md | 8 +
bin/pulsar | 6 +
distribution/server/build.gradle.kts | 2 +
gradle/libs.versions.toml | 4 +-
.../authorization/AuthorizationProvider.java | 28 ++
.../broker/authorization/AuthorizationService.java | 24 ++
.../authorization/AuthorizationServiceTest.java | 15 +
.../authorization/MockAuthorizationProvider.java | 6 +
.../org/apache/pulsar/broker/PulsarService.java | 5 +-
.../apache/pulsar/broker/admin/AdminResource.java | 3 +-
.../broker/admin/impl/PersistentTopicsBase.java | 39 ++-
.../pulsar/broker/lookup/TopicLookupBase.java | 2 +-
.../pulsar/broker/service/BrokerService.java | 117 +++++--
.../apache/pulsar/broker/service/ServerCnx.java | 90 ++++-
.../broker/service/scalable/DagWatchSession.java | 11 +-
.../pulsar/broker/web/PulsarWebResource.java | 28 ++
.../apache/pulsar/broker/PulsarServiceTest.java | 9 +
.../pulsar/broker/service/ServerCnxTest.java | 52 ++-
.../service/scalable/DagWatchSessionTest.java | 6 +-
.../api/TopicAutoCreationAuthorizationTest.java | 373 +++++++++++++++++++++
pulsar-functions/instance/build.gradle.kts | 2 +
.../functions/instance/JavaInstanceRunnable.java | 25 +-
.../pulsar/functions/instance/LogAppender.java | 62 ++++
.../pulsar/functions/instance/LogAppenderTest.java | 202 +++++++++++
...lsarLedgerUnderreplicationManagerCloseTest.java | 23 +-
.../ProxyTopicAutoCreationAuthorizationTest.java | 182 ++++++++++
.../apache/pulsar/websocket/WebSocketService.java | 11 +-
.../WebSocketTlsConfigurationStoreTest.java | 90 +++++
.../src/test/resources/pulsar-websocket.xml | 1 +
29 files changed, 1348 insertions(+), 78 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicAutoCreationAuthorizationTest.java
create mode 100644 pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/LogAppenderTest.java
create mode 100644 pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTopicAutoCreationAuthorizationTest.java
create mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/websocket/WebSocketTlsConfigurationStoreTest.java
(pulsar) 03/05: [fix][test] Fix flaky PulsarLedgerUnderreplicationManagerCloseTest.testCloseAggregatesFailuresAndRetriesOnlyUnreleasedLocks (#27013)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit bb0369c5224c7761234c0ab0210ce4d5532702d1
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Oct 20 20:03:43 2026 +0300
[fix][test] Fix flaky PulsarLedgerUnderreplicationManagerCloseTest.testCloseAggregatesFailuresAndRetriesOnlyUnreleasedLocks (#27013)
(cherry picked from commit 5307d760114dd23528dd33c16b5446f93de37abb)
---
...lsarLedgerUnderreplicationManagerCloseTest.java | 23 ++++++++++++++--------
1 file changed, 15 insertions(+), 8 deletions(-)
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java
index e41f12942eac..063054e30fba 100644
--- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java
@@ -30,6 +30,7 @@ import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -139,25 +140,31 @@ public class PulsarLedgerUnderreplicationManagerCloseTest extends BaseMetadataSt
try (var owner = newManager(observed, root);
var other = newManager(store, root)) {
acquireLedgers(owner);
+ // Deletes are issued from the store's callback threads in the order the ownership reads complete.
List<String> attemptedPaths = new ArrayList<>();
List<MetadataStoreException> failures = List.of(new MetadataStoreException("first failure"),
new MetadataStoreException("second failure"));
doAnswer(invocation -> {
String path = invocation.getArgument(0);
- attemptedPaths.add(path);
- if (attemptedPaths.size() <= failures.size()) {
- return CompletableFuture.failedFuture(failures.get(attemptedPaths.size() - 1));
+ int attempt;
+ synchronized (attemptedPaths) {
+ attempt = attemptedPaths.size();
+ attemptedPaths.add(path);
+ }
+ if (attempt < failures.size()) {
+ return CompletableFuture.failedFuture(failures.get(attempt));
}
return store.delete(path, invocation.getArgument(1));
}).when(observed).delete(anyString(), any());
+ // close() aggregates failures in held-lock order, which is independent of the delete order.
assertThatThrownBy(owner::close).isInstanceOf(UnavailableException.class)
.satisfies(error -> {
- assertThat(error.getCause()).isInstanceOf(ExecutionException.class)
- .hasCause(failures.get(0));
+ assertThat(error.getCause()).isInstanceOf(ExecutionException.class);
assertThat(error.getSuppressed()).hasSize(1);
- assertThat(error.getSuppressed()[0]).isInstanceOf(ExecutionException.class)
- .hasCause(failures.get(1));
+ assertThat(error.getSuppressed()[0]).isInstanceOf(ExecutionException.class);
+ assertThat(List.of(error.getCause().getCause(), error.getSuppressed()[0].getCause()))
+ .containsExactlyInAnyOrderElementsOf(failures);
});
assertThat(attemptedPaths).hasSize(LEDGERS.size());
for (long ledgerId : LEDGERS) {
@@ -166,7 +173,7 @@ public class PulsarLedgerUnderreplicationManagerCloseTest extends BaseMetadataSt
}
long reacquired = other.pollLedgerToRereplicate();
assertThat(reacquired).isIn(LEDGERS);
- List<String> retryPaths = new ArrayList<>();
+ List<String> retryPaths = new CopyOnWriteArrayList<>();
doAnswer(invocation -> {
String path = invocation.getArgument(0);
retryPaths.add(path);
(pulsar) 02/05: [improve][broker] Enable Log4j2 async loggers again (#26979)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ee9a0acd6f50a45f6fdc52d587dc3c0fe23b4c77
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Tue Oct 20 09:33:27 2026 -0700
[improve][broker] Enable Log4j2 async loggers again (#26979)
(cherry picked from commit 69683f49981e2abe7350c3b85d82566c9c719cdb)
---
bin/pulsar | 6 +
distribution/server/build.gradle.kts | 2 +
gradle/libs.versions.toml | 4 +-
pulsar-functions/instance/build.gradle.kts | 2 +
.../functions/instance/JavaInstanceRunnable.java | 25 ++-
.../pulsar/functions/instance/LogAppender.java | 62 +++++++
.../pulsar/functions/instance/LogAppenderTest.java | 202 +++++++++++++++++++++
7 files changed, 296 insertions(+), 7 deletions(-)
diff --git a/bin/pulsar b/bin/pulsar
index d7513c654247..9ca60ec9b8d9 100755
--- a/bin/pulsar
+++ b/bin/pulsar
@@ -339,6 +339,12 @@ OPTS="-cp $PULSAR_CLASSPATH $OPTS"
# -Dio.netty.leakDetection.level=simple in PULSAR_EXTRA_OPTS (BOOKIE_EXTRA_OPTS for bookies), override this.
OPTS="-Dio.netty.leakDetection.level=disabled $OPTS"
+# Log4j2 async loggers: the threads that log hand their events to a background thread instead of writing them, so
+# that writing the logs doesn't stall them, see https://logging.apache.org/log4j/2.x/manual/async.html.
+# To log synchronously, set -Dlog4j2.contextSelector=org.apache.logging.log4j.core.selector.ClassLoaderContextSelector
+# in PULSAR_EXTRA_OPTS (BOOKIE_EXTRA_OPTS for bookies), which come later on the command line and override this.
+OPTS="-Dlog4j2.contextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector $OPTS"
+
if [ $COMMAND == "bookie" ]; then
# Pass BOOKIE_EXTRA_OPTS option defined in bkenv.sh
OPTS="$OPTS $BOOKIE_MEM $BOOKIE_GC $BOOKIE_GC_LOG $BOOKIE_EXTRA_OPTS"
diff --git a/distribution/server/build.gradle.kts b/distribution/server/build.gradle.kts
index 944ac719bcae..4b19c700721c 100644
--- a/distribution/server/build.gradle.kts
+++ b/distribution/server/build.gradle.kts
@@ -125,6 +125,8 @@ dependencies {
distLib(libs.log4j.layout.template.json)
distLib(libs.log4j.slf4j2.impl)
distLib(libs.log4j.jul)
+ // The async loggers that bin/pulsar enables need the LMAX Disruptor
+ distLib(libs.disruptor)
distLib(libs.simpleclient.log4j2)
// Metrics
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 5fd19728518c..1dfee261c453 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -144,7 +144,8 @@ netty-reactive-streams = "2.0.20"
# 9.2.0+ is required for CronType.SPRING53 (Spring 5.3+ cron syntax)
cron-utils = "9.2.1"
failsafe = "3.3.2"
-disruptor = "3.4.3"
+# The LMAX Disruptor of log4j2's async loggers, which bin/pulsar enables
+disruptor = "4.0.0"
ant = "1.10.17"
# guice is held at 5.1.0 to match what TestNG 7.x declares; Guice 7 switches from
# javax.inject to jakarta.inject. jclouds is the only consumer that needs Guice 7 and
@@ -229,6 +230,7 @@ log4j-web = { module = "org.apache.logging.log4j:log4j-web", version.ref = "log4
log4j-layout-template-json = { module = "org.apache.logging.log4j:log4j-layout-template-json", version.ref = "log4j2" }
log4j-slf4j2-impl = { module = "org.apache.logging.log4j:log4j-slf4j2-impl", version.ref = "log4j2" }
log4j-jul = { module = "org.apache.logging.log4j:log4j-jul", version.ref = "log4j2" }
+disruptor = { module = "com.lmax:disruptor", version.ref = "disruptor" }
# slog
slog = { module = "io.github.merlimat.slog:slog", version.ref = "slog" }
# Lombok
diff --git a/pulsar-functions/instance/build.gradle.kts b/pulsar-functions/instance/build.gradle.kts
index e099acbf0d69..ece2e73b4ee2 100644
--- a/pulsar-functions/instance/build.gradle.kts
+++ b/pulsar-functions/instance/build.gradle.kts
@@ -69,6 +69,8 @@ dependencies {
// Main code only touches com.google.protobuf reflectively (protobuf schema detection for user
// functions); tests exercise that path with concrete protobuf types.
testImplementation(libs.protobuf.java)
+ // Log4j2's async loggers, which the broker and the functions worker use, for the log topic appender's tests
+ testImplementation(libs.disruptor)
}
tasks.withType<Test> {
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
index 7506e0ffd5b0..ec5a848e7bca 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -128,6 +128,9 @@ import org.slf4j.LoggerFactory;
@CustomLog
public class JavaInstanceRunnable implements AutoCloseable, Runnable {
+ // How long the close waits for the log topic's appender to receive the events logged before it
+ private static final long LOG_TOPIC_DRAIN_TIMEOUT_MS = 5000;
+
private final InstanceConfig instanceConfig;
// input topic consumer & output topic producer
@@ -715,8 +718,8 @@ public class JavaInstanceRunnable implements AutoCloseable, Runnable {
producerCache.close();
if (logAppender != null) {
- removeLogTopicAppender(LoggerContext.getContext());
- removeLogTopicAppender(LoggerContext.getContext(false));
+ removeLogTopicAppender(LoggerContext.getContext(), logAppender);
+ removeLogTopicAppender(LoggerContext.getContext(false), logAppender);
logAppender.stop();
logAppender = null;
}
@@ -860,12 +863,13 @@ public class JavaInstanceRunnable implements AutoCloseable, Runnable {
? new LogAppender(clientV5, logTopic, fqn, instanceConfig.getInstanceName())
: new LogAppender(client, logTopic, fqn, instanceConfig.getInstanceName());
logAppender.start();
- setupLogTopicAppender(LoggerContext.getContext());
- setupLogTopicAppender(LoggerContext.getContext(false));
+ setupLogTopicAppender(LoggerContext.getContext(), logAppender);
+ setupLogTopicAppender(LoggerContext.getContext(false), logAppender);
}
}
- private void setupLogTopicAppender(LoggerContext context) {
+ @VisibleForTesting
+ static void setupLogTopicAppender(LoggerContext context, LogAppender logAppender) {
Configuration config = context.getConfiguration();
config.addAppender(logAppender);
for (final LoggerConfig loggerConfig : config.getLoggers().values()) {
@@ -875,7 +879,16 @@ public class JavaInstanceRunnable implements AutoCloseable, Runnable {
context.updateLoggers();
}
- private void removeLogTopicAppender(LoggerContext context) {
+ @VisibleForTesting
+ static void removeLogTopicAppender(LoggerContext context, LogAppender logAppender) {
+ // With async loggers, the events logged before the close may still be queued: let them reach the log topic
+ // before the loggers stop handing events to the appender
+ if (!logAppender.drain(context, LOG_TOPIC_DRAIN_TIMEOUT_MS)) {
+ log.warn()
+ .attr("function", logAppender.getName())
+ .attr("timeoutMs", LOG_TOPIC_DRAIN_TIMEOUT_MS)
+ .log("Removing the log topic appender before it received every event logged before the close");
+ }
Configuration config = context.getConfiguration();
for (final LoggerConfig loggerConfig : config.getLoggers().values()) {
loggerConfig.removeAppender(logAppender.getName());
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/LogAppender.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/LogAppender.java
index 673389246ea9..a263843383d3 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/LogAppender.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/LogAppender.java
@@ -21,13 +21,25 @@ package org.apache.pulsar.functions.instance;
import java.io.Serializable;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.Marker;
+import org.apache.logging.log4j.MarkerManager;
import org.apache.logging.log4j.core.Appender;
import org.apache.logging.log4j.core.ErrorHandler;
import org.apache.logging.log4j.core.Layout;
import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
import org.apache.logging.log4j.core.appender.DefaultErrorHandler;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.config.LoggerConfig;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
@@ -46,6 +58,8 @@ public class LogAppender implements Appender {
private static final String LOG_LEVEL = "loglevel";
private static final String INSTANCE = "instance";
private static final String FQN = "fqn";
+ // The marker of the events that drain() logs, which end its wait instead of going to the log topic
+ private static final Marker DRAIN_MARKER = MarkerManager.getMarker("PULSAR_FUNCTION_LOG_TOPIC_DRAIN");
private PulsarClient pulsarClient;
private Supplier<org.apache.pulsar.client.api.v5.PulsarClient> pulsarClientV5;
@@ -55,6 +69,8 @@ public class LogAppender implements Appender {
private State state;
private ErrorHandler errorHandler;
private Producer<byte[]> producer;
+ // The waits of drain(), by the message of the event that each one waits for
+ private final Map<String, CompletableFuture<Void>> drains = new ConcurrentHashMap<>();
public LogAppender(PulsarClient pulsarClient, String logTopic, String fqn, String instance) {
this.pulsarClient = pulsarClient;
@@ -75,6 +91,13 @@ public class LogAppender implements Appender {
@Override
public void append(LogEvent logEvent) {
+ if (DRAIN_MARKER.equals(logEvent.getMarker())) {
+ CompletableFuture<Void> drained = drains.get(logEvent.getMessage().getFormattedMessage());
+ if (drained != null) {
+ drained.complete(null);
+ }
+ return;
+ }
producer.newMessage()
.value(logEvent.getMessage().getFormattedMessage().getBytes(StandardCharsets.UTF_8))
.property(LOG_LEVEL, logEvent.getLevel().name())
@@ -83,6 +106,45 @@ public class LogAppender implements Appender {
.sendAsync();
}
+ /**
+ * Waits until this appender has received the events that the context's loggers accepted before this call, for at
+ * most the timeout, and returns whether it did.
+ *
+ * <p>With async loggers, the context's background thread hands each event to the appenders attached when it gets
+ * to the event, so an event still queued when the appender is detached never reaches the log topic. The thread
+ * handles the events in order: this logs an event that only this appender receives, through a logger of its own,
+ * and waits for it. With synchronous loggers, the event arrives at once.
+ */
+ boolean drain(LoggerContext context, long timeoutMillis) {
+ String loggerName = LogAppender.class.getName() + ".drain." + fqn + "." + instance;
+ Configuration config = context.getConfiguration();
+ // The logger's configuration stays, without appenders, after the drain: an event that the background thread
+ // handles after the timeout must not reach the configuration's other appenders through the root logger
+ if (!config.getLoggers().containsKey(loggerName)) {
+ config.addLogger(loggerName, new LoggerConfig(loggerName, Level.ALL, false));
+ }
+ LoggerConfig loggerConfig = config.getLoggers().get(loggerName);
+ loggerConfig.addAppender(this, null, null);
+ context.updateLoggers();
+ String id = UUID.randomUUID().toString();
+ CompletableFuture<Void> drained = new CompletableFuture<>();
+ drains.put(id, drained);
+ try {
+ // At the most severe level, which a queue-full policy that discards events keeps
+ context.getLogger(loggerName).log(Level.FATAL, DRAIN_MARKER, id);
+ drained.get(timeoutMillis, TimeUnit.MILLISECONDS);
+ return true;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ } catch (ExecutionException | TimeoutException e) {
+ return false;
+ } finally {
+ drains.remove(id);
+ loggerConfig.removeAppender(getName());
+ }
+ }
+
@Override
public String getName() {
return fqn;
diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/LogAppenderTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/LogAppenderTest.java
new file mode 100644
index 000000000000..fa2400124a4a
--- /dev/null
+++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/LogAppenderTest.java
@@ -0,0 +1,202 @@
+/*
+ * 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.functions.instance;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.RETURNS_SELF;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.Logger;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.async.AsyncLoggerContext;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.config.Property;
+import org.apache.logging.log4j.core.config.builder.api.ConfigurationBuilder;
+import org.apache.logging.log4j.core.config.builder.api.ConfigurationBuilderFactory;
+import org.apache.logging.log4j.core.config.builder.impl.BuiltConfiguration;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * The log topic's appender when a function instance closes: the events logged before the close reach the log topic,
+ * also when async loggers hand them to the appender after the close started.
+ */
+public class LogAppenderTest {
+
+ @Test
+ public void testEventsQueuedBeforeTheCloseReachTheLogTopic() throws Exception {
+ RecordingAppender recording = new RecordingAppender();
+ AsyncLoggerContext context = start(new AsyncLoggerContext("log-appender-async-test"), recording);
+ List<String> published = new CopyOnWriteArrayList<>();
+ try {
+ LogAppender logAppender = newLogAppender(published);
+ JavaInstanceRunnable.setupLogTopicAppender(context, logAppender);
+ Logger logger = context.getLogger("function");
+ // Hold the background thread on an earlier event, so that the last event waits in the queue
+ logger.info("block");
+ assertTrue(recording.blocked.await(10, TimeUnit.SECONDS));
+ logger.info("last");
+
+ Thread close = new Thread(() -> {
+ JavaInstanceRunnable.removeLogTopicAppender(context, logAppender);
+ logAppender.stop();
+ });
+ close.start();
+ // Release the background thread once the close waits for it, or has detached the appender without waiting
+ Awaitility.await().until(() -> close.getState() == Thread.State.TIMED_WAITING
+ || close.getState() == Thread.State.TERMINATED);
+ recording.release.countDown();
+ close.join(TimeUnit.SECONDS.toMillis(10));
+
+ assertFalse(close.isAlive());
+ assertEquals(published, List.of("block", "last"));
+ assertDetached(context, logAppender);
+ } finally {
+ recording.release.countDown();
+ context.stop(10, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void testDrainWithSynchronousLoggers() throws Exception {
+ LoggerContext context = start(new LoggerContext("log-appender-sync-test"), new RecordingAppender());
+ List<String> published = new CopyOnWriteArrayList<>();
+ try {
+ LogAppender logAppender = newLogAppender(published);
+ JavaInstanceRunnable.setupLogTopicAppender(context, logAppender);
+ context.getLogger("function").info("last");
+
+ assertTrue(logAppender.drain(context, 0));
+ JavaInstanceRunnable.removeLogTopicAppender(context, logAppender);
+ logAppender.stop();
+
+ assertEquals(published, List.of("last"));
+ assertDetached(context, logAppender);
+ } finally {
+ context.stop(10, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void testDrainGivesUpAfterTheTimeout() throws Exception {
+ RecordingAppender recording = new RecordingAppender();
+ AsyncLoggerContext context = start(new AsyncLoggerContext("log-appender-timeout-test"), recording);
+ List<String> published = new CopyOnWriteArrayList<>();
+ try {
+ LogAppender logAppender = newLogAppender(published);
+ JavaInstanceRunnable.setupLogTopicAppender(context, logAppender);
+ Logger logger = context.getLogger("function");
+ logger.info("block");
+ assertTrue(recording.blocked.await(10, TimeUnit.SECONDS));
+
+ assertFalse(logAppender.drain(context, 100));
+
+ // The drain's event, which the background thread handles after the timeout, reaches no other appender
+ recording.release.countDown();
+ logger.info("after");
+ Awaitility.await().until(() -> recording.received.contains("after"));
+ assertEquals(recording.received, List.of("block", "after"));
+ assertEquals(published, List.of("block", "after"));
+ } finally {
+ recording.release.countDown();
+ context.stop(10, TimeUnit.SECONDS);
+ }
+ }
+
+ private static <T extends LoggerContext> T start(T context, RecordingAppender recording) {
+ ConfigurationBuilder<BuiltConfiguration> builder = ConfigurationBuilderFactory.newConfigurationBuilder();
+ builder.setConfigurationName(context.getName());
+ builder.add(builder.newRootLogger(Level.INFO));
+ context.start(builder.build(false));
+ recording.start();
+ context.getConfiguration().getRootLogger().addAppender(recording, null, null);
+ context.updateLoggers();
+ return context;
+ }
+
+ // An appender for the log topic, whose producer adds the messages that it publishes to the list
+ @SuppressWarnings("unchecked")
+ private static LogAppender newLogAppender(List<String> published) throws Exception {
+ TypedMessageBuilder<byte[]> message = mock(TypedMessageBuilder.class, RETURNS_SELF);
+ when(message.value(any())).thenAnswer(invocation -> {
+ published.add(new String((byte[]) invocation.getArgument(0), StandardCharsets.UTF_8));
+ return message;
+ });
+ when(message.sendAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ Producer<byte[]> producer = mock(Producer.class);
+ when(producer.newMessage()).thenReturn(message);
+ when(producer.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ ProducerBuilder<byte[]> producerBuilder = mock(ProducerBuilder.class, RETURNS_SELF);
+ when(producerBuilder.create()).thenReturn(producer);
+ PulsarClient client = mock(PulsarClient.class);
+ when(client.newProducer()).thenReturn(producerBuilder);
+ LogAppender logAppender = new LogAppender(client, "log-topic", "tenant/namespace/function", "0");
+ logAppender.start();
+ return logAppender;
+ }
+
+ private static void assertDetached(LoggerContext context, LogAppender logAppender) {
+ Configuration config = context.getConfiguration();
+ assertFalse(config.getRootLogger().getAppenders().containsKey(logAppender.getName()));
+ assertTrue(config.getLoggers().values().stream()
+ .noneMatch(loggerConfig -> loggerConfig.getAppenders().containsKey(logAppender.getName())));
+ }
+
+ // Records the messages of the events that it receives, and holds the thread that hands it the event "block" until
+ // released
+ private static final class RecordingAppender extends AbstractAppender {
+ private final List<String> received = new CopyOnWriteArrayList<>();
+ private final CountDownLatch blocked = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ RecordingAppender() {
+ super("recording", null, null, true, Property.EMPTY_ARRAY);
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ String message = event.getMessage().getFormattedMessage();
+ received.add(message);
+ if (message.equals("block")) {
+ blocked.countDown();
+ try {
+ release.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+ }
+}
(pulsar) 04/05: [improve][doc] Note in CODING.md that a delayedExecutor() task cannot be taken back (#26988)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4b8e30d188ecf92eada8775771e4e7ae0ee7ad5b
Author: SeungTaek Lim <na...@gmail.com>
AuthorDate: Wed Oct 21 05:39:31 2026 +0900
[improve][doc] Note in CODING.md that a delayedExecutor() task cannot be taken back (#26988)
(cherry picked from commit 0da9e8b970d6d7ebc17cc5704db019d9594c4e1a)
---
CODING.md | 8 ++++++++
1 file changed, 8 insertions(+)
diff --git a/CODING.md b/CODING.md
index 99adf8f26b77..15ffdd045d63 100644
--- a/CODING.md
+++ b/CODING.md
@@ -99,6 +99,14 @@ Pulsar relies heavily on `CompletableFuture`; prefer it over `ListenableFuture`
- **`org.apache.pulsar.common.semaphore.AsyncSemaphoreImpl`** — a non-blocking semaphore with a
per-operation cost that queues callers instead of failing when the limit is reached. Preferred over
`ConcurrencyReducer` for request-driven cases that need a timeout on permit acquisition.
+- **A task given to `CompletableFuture.delayedExecutor()` cannot be taken back.** It sits in the JDK's
+ delayer queue until its delay elapses and keeps everything its lambda captured (typically `this`)
+ reachable until then; cancelling a future built on it (`runAsync(task, delayedExecutor(...))`) only
+ skips the run. A `ScheduledThreadPoolExecutor` configured with `setRemoveOnCancelPolicy(true)` does
+ drop a cancelled task, but a task queued there is discarded without notice by `shutdownNow()`, so
+ neither is right for every timer. When using the delayer: check a stop flag when the task runs, keep
+ the delay or the captured state small, and when the retention matters, capture a reference that the
+ stop can clear instead of the object itself.
## Testing conventions
(pulsar) 05/05: [improve][broker] Let authorization providers restrict topic auto-creation (#17411)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 48f850be5f1ff2d5ca7e23453efa2ed913d80b51
Author: kannar <ka...@gmail.com>
AuthorDate: Tue Oct 20 23:35:21 2026 +0200
[improve][broker] Let authorization providers restrict topic auto-creation (#17411)
(cherry picked from commit 0769a0ff360b51e40fafe3012981ec06227c6665)
---
.../authorization/AuthorizationProvider.java | 28 ++
.../broker/authorization/AuthorizationService.java | 24 ++
.../authorization/AuthorizationServiceTest.java | 15 +
.../authorization/MockAuthorizationProvider.java | 6 +
.../apache/pulsar/broker/admin/AdminResource.java | 3 +-
.../broker/admin/impl/PersistentTopicsBase.java | 39 ++-
.../pulsar/broker/lookup/TopicLookupBase.java | 2 +-
.../pulsar/broker/service/BrokerService.java | 117 +++++--
.../apache/pulsar/broker/service/ServerCnx.java | 90 ++++-
.../broker/service/scalable/DagWatchSession.java | 11 +-
.../pulsar/broker/web/PulsarWebResource.java | 28 ++
.../pulsar/broker/service/ServerCnxTest.java | 52 ++-
.../service/scalable/DagWatchSessionTest.java | 6 +-
.../api/TopicAutoCreationAuthorizationTest.java | 373 +++++++++++++++++++++
.../ProxyTopicAutoCreationAuthorizationTest.java | 182 ++++++++++
15 files changed, 916 insertions(+), 60 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
index f3cce0a2a5a2..7ca7945d7b1d 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
@@ -345,6 +345,34 @@ public interface AuthorizationProvider extends Closeable {
+ "provider you are using."));
}
+ /**
+ * Check if a given <tt>role</tt> may have a topic created automatically, when it produces to, subscribes to or
+ * looks up a topic that does not exist and the auto topic creation settings allow creating it.
+ *
+ * <p>The default allows it, so the auto topic creation settings alone decide, as they did before this check
+ * existed. A provider overrides it to restrict which roles may trigger topic auto-creation, for example by
+ * checking {@link NamespaceOperation#CREATE_TOPIC}.
+ *
+ * <p>It is only called when the settings would create the topic from nothing: not for a topic the broker
+ * re-materializes from existing metadata (a partition of an existing partitioned topic, an active segment of a
+ * scalable topic), nor for system topics, and only once the broker found that the topic does not exist, in this
+ * cluster or, for partitioned-topic metadata, in a remote one. The exception is an HTTP request for a
+ * non-persistent topic, whose existence the broker does not check there, so it must not assume the topic is
+ * missing. It should be cheap. A refusal only prevents creating a missing topic, and a future that completes
+ * exceptionally fails the client request. The broker's own clients, such as geo-replication, go through it too
+ * with their role, so a provider should keep allowing super users.
+ *
+ * @param topic topic name
+ * @param role role name
+ * @param authData authenticated data
+ * @return CompletableFuture<Boolean>
+ */
+ default CompletableFuture<Boolean> allowTopicAutoCreationAsync(TopicName topic,
+ String role,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(true);
+ }
+
/**
* Check if a given <tt>role</tt> is allowed to execute a given topic <tt>operation</tt> on topic's <tt>policy</tt>.
*
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
index 6f9bdd14771b..f63318963611 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
@@ -964,6 +964,30 @@ public class AuthorizationService {
}
}
+ /**
+ * Check if a client may have a topic created automatically, see
+ * {@link AuthorizationProvider#allowTopicAutoCreationAsync}. Through a proxy, both the proxy role and the
+ * original principal must be allowed, as for topic operations.
+ */
+ public CompletableFuture<Boolean> allowTopicAutoCreationAsync(TopicName topicName,
+ String originalRole,
+ String role,
+ AuthenticationDataSource originalAuthData,
+ AuthenticationDataSource authData) {
+ if (!this.conf.isAuthorizationEnabled()) {
+ return CompletableFuture.completedFuture(true);
+ }
+ if (!isValidOriginalPrincipal(role, originalRole, originalAuthData)) {
+ return CompletableFuture.completedFuture(false);
+ }
+ if (isProxyRole(role) && !isWebsocketPrinciple(originalRole)) {
+ return provider.allowTopicAutoCreationAsync(topicName, role, authData)
+ .thenCombine(provider.allowTopicAutoCreationAsync(topicName, originalRole, originalAuthData),
+ (isRoleAllowed, isOriginalAllowed) -> isRoleAllowed && isOriginalAllowed);
+ }
+ return provider.allowTopicAutoCreationAsync(topicName, role, authData);
+ }
+
public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topicName,
TopicOperation operation,
String originalRole,
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java
index f90c98bfabbe..b2e113598056 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java
@@ -209,6 +209,21 @@ public class AuthorizationServiceTest {
checkResult(shouldPass, isAuthorized);
}
+ @Test(dataProvider = "roles")
+ public void testTopicAutoCreationAsync(String role, String originalRole, boolean shouldPass) throws Exception {
+ boolean isAllowed = authorizationService.allowTopicAutoCreationAsync(TopicName.get("topic"),
+ originalRole, role, null, null).get();
+ checkResult(shouldPass, isAllowed);
+ }
+
+ @Test
+ public void testTopicAutoCreationAllowedByDefault() throws Exception {
+ // The default provider does not override the check, so this runs the interface's default method.
+ AuthorizationProvider provider = new PulsarAuthorizationProvider();
+ assertThat(provider.allowTopicAutoCreationAsync(TopicName.get("topic"), "any-role", null).get())
+ .as("auto-creation allowed by the default implementation").isTrue();
+ }
+
@Test(dataProvider = "roles")
public void testNamespacePolicyOperationAsync(String role, String originalRole, boolean shouldPass)
throws Exception {
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MockAuthorizationProvider.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MockAuthorizationProvider.java
index 9f9cd92cd0a7..1dcb378b5edb 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MockAuthorizationProvider.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MockAuthorizationProvider.java
@@ -98,6 +98,12 @@ public class MockAuthorizationProvider implements AuthorizationProvider {
return shouldPass(role);
}
+ @Override
+ public CompletableFuture<Boolean> allowTopicAutoCreationAsync(TopicName topic, String role,
+ AuthenticationDataSource authData) {
+ return shouldPass(role);
+ }
+
@Override
public CompletableFuture<Boolean> allowTopicPolicyOperationAsync(TopicName topic, String role, PolicyName policy,
PolicyOperation operation,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index 84a548ddc5ce..7b90a65ad1db 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -485,7 +485,8 @@ public abstract class AdminResource extends PulsarWebResource {
.thenCompose(__ -> {
if (checkAllowAutoCreation) {
return pulsar().getBrokerService()
- .fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName);
+ .fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName,
+ newTopicCheck(topicName));
} else {
return pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName);
}
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 0eaa171770e8..7440e8fc6ebe 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
@@ -691,9 +691,18 @@ public class PersistentTopicsBase extends AdminResource {
// However, when checkAllowAutoCreation is true, the client will create the topic if
// it doesn't exist. In this case, `partitions == 0` means the automatically created topic
// is a non-partitioned topic so we shouldn't check if the topic exists.
- return pulsar().getBrokerService().isAllowAutoTopicCreationAsync(topicName)
- .thenCompose(brokerAllowAutoTopicCreation -> {
- if (checkAllowAutoCreation && brokerAllowAutoTopicCreation) {
+ if (!checkAllowAutoCreation) {
+ return internalCheckNonPartitionedTopicExists(topicName).thenApply(__ -> metadata);
+ }
+ // Only ask whether this client may create the topic when it does not exist. The existence of
+ // a non-persistent topic is not checked, as that may need a request to its owner broker.
+ return (topicName.isPersistent()
+ ? pulsar().getNamespaceService().checkNonPartitionedTopicExists(topicName)
+ : CompletableFuture.completedFuture(false))
+ .thenCompose(exists -> exists ? CompletableFuture.completedFuture(true)
+ : isAllowAutoTopicCreationAsync(topicName))
+ .thenCompose(allowAutoTopicCreation -> {
+ if (allowAutoTopicCreation) {
// Whether it exists or not, auto create a non-partitioned topic by client.
return CompletableFuture.completedFuture(metadata);
} else {
@@ -2612,9 +2621,8 @@ public class PersistentTopicsBase extends AdminResource {
validateTopicOwnershipAsync(topicName, authoritative)
.thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.SUBSCRIBE, subscriptionName))
- .thenCompose(__ -> pulsar().getBrokerService().isAllowAutoTopicCreationAsync(topicName))
- .thenCompose(isAllowAutoTopicCreation -> pulsar().getBrokerService()
- .getTopic(topicName.toString(), isAllowAutoTopicCreation))
+ .thenCompose(__ -> pulsar().getBrokerService()
+ .getTopicCreatingIfAllowed(topicName, newTopicCheck(topicName)))
.thenApply(optTopic -> {
if (optTopic.isPresent()) {
return optTopic.get();
@@ -4789,8 +4797,10 @@ public class PersistentTopicsBase extends AdminResource {
authorizationFuture.thenCompose(__ ->
checkLocalOrGetPeerReplicationCluster(pulsar, topicName.getNamespaceObject(),
SystemTopicNames.isSystemTopic(topicName)))
- .thenCompose(res ->
- pulsar.getBrokerService().fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName))
+ .thenCompose(res -> pulsar.getBrokerService().fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(
+ topicName, () -> pulsar.getBrokerService().getAuthorizationService()
+ .allowTopicAutoCreationAsync(topicName, originalPrincipal, clientAppId,
+ originalAuthenticationData, authenticationData)))
.thenAccept(metadata -> {
log.debug()
.attr("topic", topicName)
@@ -4810,6 +4820,17 @@ public class PersistentTopicsBase extends AdminResource {
*/
public static CompletableFuture<PartitionedTopicMetadata> unsafeGetPartitionedTopicMetadataAsync(
PulsarService pulsar, TopicName topicName) {
+ return unsafeGetPartitionedTopicMetadataAsync(pulsar, topicName,
+ () -> CompletableFuture.completedFuture(true));
+ }
+
+ /**
+ * Like {@link #unsafeGetPartitionedTopicMetadataAsync(PulsarService, TopicName)}, but auto-creating a topic from
+ * nothing is also subject to {@code newTopicCheck}, see
+ * {@code BrokerService#fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(TopicName, Supplier)}.
+ */
+ public static CompletableFuture<PartitionedTopicMetadata> unsafeGetPartitionedTopicMetadataAsync(
+ PulsarService pulsar, TopicName topicName, Supplier<CompletableFuture<Boolean>> newTopicCheck) {
CompletableFuture<PartitionedTopicMetadata> metadataFuture = new CompletableFuture<>();
// validates global-namespace contains local/peer cluster: if peer/local cluster present then lookup can
@@ -4821,7 +4842,7 @@ public class PersistentTopicsBase extends AdminResource {
// If we don't do this, it can prevent namespace deletion due to inaccessible readers.
checkLocalOrGetPeerReplicationCluster(pulsar, topicName.getNamespaceObject(), isSystemTopic(topicName))
.thenCompose(res -> pulsar.getBrokerService()
- .fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName))
+ .fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName, newTopicCheck))
.thenAccept(metadata -> {
LOG.debug()
.attr("topic", topicName)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
index 174c96bd8054..b07d0089d243 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
@@ -89,7 +89,7 @@ public class TopicLookupBase extends PulsarWebResource {
if (exists) {
return CompletableFuture.completedFuture(true);
}
- return pulsar().getBrokerService().isAllowAutoTopicCreationAsync(topicName);
+ return isAllowAutoTopicCreationAsync(topicName);
});
})
.thenCompose(exist -> {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 47030c507823..610ab2b245f7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -83,6 +83,7 @@ import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Consumer;
import java.util.function.Predicate;
+import java.util.function.Supplier;
import lombok.AccessLevel;
import lombok.CustomLog;
import lombok.Getter;
@@ -231,6 +232,8 @@ public class BrokerService implements Closeable {
"futureWithDeadline(...)");
private static final double GRACEFUL_SHUTDOWN_QUIET_PERIOD_RATIO_OF_TOTAL_TIMEOUT = 0.25d;
private static final double GRACEFUL_SHUTDOWN_TIMEOUT_RATIO_OF_TOTAL_TIMEOUT = 0.5d;
+ private static final Supplier<CompletableFuture<Boolean>> ALLOW_NEW_TOPIC =
+ () -> CompletableFuture.completedFuture(true);
private static final Histogram backlogQuotaCheckDuration = Histogram.build()
.name("pulsar_storage_backlog_quota_check_duration_seconds")
@@ -1313,6 +1316,40 @@ public class BrokerService implements Closeable {
return getTopic(topic, createIfMissing, null);
}
+ /**
+ * Loads the topic and, when it does not exist, creates it if
+ * {@link #isAllowAutoTopicCreationAsync(TopicName, Supplier)} allows it with {@code newTopicCheck}. The check is
+ * only made for a topic that does not exist, so loading an existing topic does not depend on it.
+ */
+ public CompletableFuture<Optional<Topic>> getTopicCreatingIfAllowed(
+ final TopicName topicName, final Supplier<CompletableFuture<Boolean>> newTopicCheck) {
+ CompletableFuture<Optional<Topic>> existingTopic = topicName.isPersistent()
+ ? getTopic(topicName.toString(), false)
+ // A non-persistent topic only exists while it is loaded; trying to load it would report a failure
+ // to topic event listeners.
+ : CompletableFuture.completedFuture(getTopicReference(topicName.toString()));
+ return existingTopic.thenCompose(optTopic -> {
+ if (optTopic.isPresent()) {
+ return CompletableFuture.completedFuture(optTopic);
+ }
+ evictMissingTopicFuture(topicName.toString());
+ return isAllowAutoTopicCreationAsync(topicName, newTopicCheck)
+ .thenCompose(isAllowed -> getTopic(topicName.toString(), isAllowed));
+ });
+ }
+
+ /**
+ * Removes the future cached for the topic when it is a completed load that found the topic missing. A load removes
+ * that future from the cache asynchronously, so creating the topic right after could otherwise get it back. The
+ * cached future may come from another load than the caller's, when two of them raced.
+ */
+ private void evictMissingTopicFuture(String topic) {
+ CompletableFuture<Optional<Topic>> cached = topics.get(topic);
+ if (cached != null && cached.isDone() && !cached.isCompletedExceptionally() && cached.join().isEmpty()) {
+ topics.remove(topic, cached);
+ }
+ }
+
public CompletableFuture<Optional<Topic>> getTopic(final String topic, boolean createIfMissing,
@Nullable Map<String, String> properties) {
return getTopic(TopicName.get(topic), createIfMissing, properties);
@@ -4151,6 +4188,17 @@ public class BrokerService implements Closeable {
public CompletableFuture<PartitionedTopicMetadata> fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(
TopicName topicName) {
+ return fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName, ALLOW_NEW_TOPIC);
+ }
+
+ /**
+ * Like {@link #fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(TopicName)}, but auto-creation is also
+ * subject to {@code newTopicCheck} when the topic would be created from nothing, see
+ * {@link #isAllowAutoTopicCreationAsync(TopicName, Supplier)}. It is only consulted once the topic is known to
+ * exist neither in this cluster nor in a remote one.
+ */
+ public CompletableFuture<PartitionedTopicMetadata> fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(
+ TopicName topicName, Supplier<CompletableFuture<Boolean>> newTopicCheck) {
if (pulsar.getNamespaceService() == null) {
return FutureUtil.failedFuture(new NamingException("namespace service is not ready"));
}
@@ -4171,7 +4219,13 @@ public class BrokerService implements Closeable {
// Try created if allowed to create a partitioned topic automatically.
return pulsar.getPulsarResources().getNamespaceResources().getPoliciesAsync(topicName.getNamespaceObject())
.thenComposeAsync(policies -> {
- return isAllowAutoTopicCreationAsync(topicName, policies).thenComposeAsync(allowed -> {
+ // Whether the settings would create the topic from nothing, in which case newTopicCheck is
+ // consulted once we know the topic does not exist in a remote cluster either.
+ AtomicBoolean isNewTopic = new AtomicBoolean();
+ return isAllowAutoTopicCreationAsync(topicName, policies, () -> {
+ isNewTopic.set(true);
+ return CompletableFuture.completedFuture(true);
+ }).thenComposeAsync(allowed -> {
// Not Allow auto-creation.
if (!allowed) {
// Do not change the original behavior, or default return a non-partitioned topic.
@@ -4188,17 +4242,24 @@ public class BrokerService implements Closeable {
}
return CompletableFuture.completedFuture(new PartitionedTopicMetadata(0));
}
+ return (isNewTopic.get() ? newTopicCheck.get()
+ : CompletableFuture.completedFuture(true)).thenCompose(isNewTopicAllowed -> {
+ if (!isNewTopicAllowed) {
+ // The same answer as when the settings forbid auto-creation.
+ return CompletableFuture.completedFuture(new PartitionedTopicMetadata(0));
+ }
- // Allow auto create non-partitioned topic.
- boolean autoCreatePartitionedTopic = pulsar.getBrokerService()
- .isDefaultTopicTypePartitioned(topicName, policies);
- if (!autoCreatePartitionedTopic || topicName.isPartitioned()) {
- return CompletableFuture.completedFuture(new PartitionedTopicMetadata(0));
- }
+ // Allow auto create non-partitioned topic.
+ boolean autoCreatePartitionedTopic = pulsar.getBrokerService()
+ .isDefaultTopicTypePartitioned(topicName, policies);
+ if (!autoCreatePartitionedTopic || topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(new PartitionedTopicMetadata(0));
+ }
- // Create partitioned metadata.
- return pulsar.getBrokerService().createDefaultPartitionedTopicAsync(topicName,
- policies);
+ // Create partitioned metadata.
+ return pulsar.getBrokerService().createDefaultPartitionedTopicAsync(topicName,
+ policies);
+ });
}).exceptionallyCompose(ex -> {
// The partitioned topic might be created concurrently.
if (ex.getCause() instanceof MetadataStoreException.AlreadyExistsException) {
@@ -4585,13 +4646,25 @@ public class BrokerService implements Closeable {
}
public CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName topicName) {
+ return isAllowAutoTopicCreationAsync(topicName, ALLOW_NEW_TOPIC);
+ }
+
+ /**
+ * Like {@link #isAllowAutoTopicCreationAsync(TopicName)}, but consults {@code newTopicCheck} too when the topic
+ * would be created from nothing, because the auto topic creation settings allow it. It is not consulted, and
+ * does not change the result, when the topic is one the broker re-materializes from existing metadata (a
+ * partition of an existing partitioned topic, an active segment of a scalable topic) or a system topic.
+ */
+ public CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(
+ final TopicName topicName, final Supplier<CompletableFuture<Boolean>> newTopicCheck) {
return pulsar.getPulsarResources().getNamespaceResources()
.getPoliciesAsync(topicName.getNamespaceObject())
- .thenCompose(policies -> isAllowAutoTopicCreationAsync(topicName, policies));
+ .thenCompose(policies -> isAllowAutoTopicCreationAsync(topicName, policies, newTopicCheck));
}
- private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName topicName,
- final Optional<Policies> policies) {
+ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(
+ final TopicName topicName, final Optional<Policies> policies,
+ final Supplier<CompletableFuture<Boolean>> newTopicCheck) {
if (policies.isPresent() && policies.get().deleted) {
log.info()
.attr("namespaceObject", topicName.getNamespaceObject())
@@ -4662,14 +4735,15 @@ public class BrokerService implements Closeable {
return scalableResources.getScalableTopicMetadataAsync(topicName.toScalableTopic())
.thenCompose(scalableMetadata -> scalableMetadata.isPresent()
? CompletableFuture.completedFuture(false)
- : isAllowAutoTopicCreationResolvedAsync(topicName, policies));
+ : isAllowAutoTopicCreationResolvedAsync(topicName, policies, newTopicCheck));
}
}
- return isAllowAutoTopicCreationResolvedAsync(topicName, policies);
+ return isAllowAutoTopicCreationResolvedAsync(topicName, policies, newTopicCheck);
}
private CompletableFuture<Boolean> isAllowAutoTopicCreationResolvedAsync(
- final TopicName topicName, final Optional<Policies> policies) {
+ final TopicName topicName, final Optional<Policies> policies,
+ final Supplier<CompletableFuture<Boolean>> newTopicCheck) {
//Other system topics can be created automatically
if (pulsar.getConfiguration().isSystemTopicEnabled() && isSystemTopic(topicName)) {
return CompletableFuture.completedFuture(true);
@@ -4688,17 +4762,18 @@ public class BrokerService implements Closeable {
// cannot re-create topic while it is being deleted
return pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
.getPartitionedTopicMetadataAsync(partitionedTopic, true)
- .thenApply(partitionedTopicMetadata -> {
+ .thenCompose(partitionedTopicMetadata -> {
if (partitionedTopicMetadata.isEmpty()) {
- return allowed;
+ return allowed ? newTopicCheck.get() : CompletableFuture.completedFuture(false);
}
if (partitionedTopicMetadata.get().deleted) {
- return false;
+ return CompletableFuture.completedFuture(false);
}
- return partitionedTopicMetadata.get().partitions > topicName.getPartitionIndex();
+ return CompletableFuture.completedFuture(
+ partitionedTopicMetadata.get().partitions > topicName.getPartitionIndex());
});
} else {
- return CompletableFuture.completedFuture(allowed);
+ return allowed ? newTopicCheck.get() : CompletableFuture.completedFuture(false);
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index 1e437b59d8f2..e1a60b419a1c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -68,6 +68,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BooleanSupplier;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import javax.naming.AuthenticationException;
import javax.net.ssl.SSLSession;
@@ -729,6 +730,52 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
}
}
+ /**
+ * Whether a topic that does not exist may be created for this client: the auto topic creation settings allow it
+ * and, when it would be created from nothing, the authorization provider lets this client trigger it. The
+ * provider is not consulted for a topic the broker re-materializes from existing metadata, such as a partition
+ * of an existing partitioned topic, see {@link BrokerService#isAllowAutoTopicCreationAsync}.
+ */
+ public CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(TopicName topicName) {
+ return service.isAllowAutoTopicCreationAsync(topicName, newTopicCheck(topicName));
+ }
+
+ /**
+ * Whether the authorization provider lets this client trigger the creation of a new topic.
+ */
+ private Supplier<CompletableFuture<Boolean>> newTopicCheck(TopicName topicName) {
+ if (!service.isAuthorizationEnabled()) {
+ return () -> CompletableFuture.completedFuture(true);
+ }
+ // The identity of the client when the request arrived, as the check may run later on another thread.
+ String role = authRole;
+ String originalRole = originalPrincipal;
+ AuthenticationDataSource authData = authenticationData;
+ AuthenticationDataSource originalData = originalAuthData != null ? originalAuthData : authData;
+ return () -> service.getAuthorizationService()
+ .allowTopicAutoCreationAsync(topicName, originalRole, role, originalData, authData)
+ .thenApply(isAllowed -> {
+ if (!isAllowed) {
+ log.debug()
+ .attr("topic", topicName)
+ .attr("role", role)
+ .attr("originalPrincipal", originalRole)
+ .log("The authorization provider refused to auto-create the topic");
+ }
+ return isAllowed;
+ });
+ }
+
+ /**
+ * Loads the topic and, when it does not exist and {@code createIfMissing} is set, creates it if
+ * {@link #isAllowAutoTopicCreationAsync} allows it, see {@link BrokerService#getTopicCreatingIfAllowed}.
+ */
+ private CompletableFuture<Optional<Topic>> getTopic(TopicName topicName, boolean createIfMissing) {
+ return createIfMissing
+ ? service.getTopicCreatingIfAllowed(topicName, newTopicCheck(topicName))
+ : service.getTopic(topicName.toString(), false);
+ }
+
@Override
protected void handleLookup(CommandLookupTopic lookupParam) {
checkArgument(state == State.Connected);
@@ -1560,13 +1607,17 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
private CompletableFuture<Void> lookupPartitionMetadata(TopicName topicName, long requestId,
boolean metadataAutoCreationEnabled) {
+ if (!metadataAutoCreationEnabled) {
+ return lookupExistingTopicMetadata(topicName, requestId);
+ }
+ // The authorization provider is only consulted later, if the topic turns out not to exist.
return FutureUtil.supplySafely(() -> getBrokerService().isAllowAutoTopicCreationAsync(topicName))
.<CompletableFuture<Void>>handle((brokerAllowAutoCreate, ex) -> {
if (ex != null) {
sendPartitionMetadataLookupError(topicName, requestId, ex);
return CompletableFuture.completedFuture(null);
}
- if (metadataAutoCreationEnabled && brokerAllowAutoCreate) {
+ if (brokerAllowAutoCreate) {
return lookupOrCreatePartitionMetadata(topicName, requestId);
}
return lookupExistingTopicMetadata(topicName, requestId);
@@ -1614,14 +1665,40 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
}
private CompletableFuture<Void> lookupOrCreatePartitionMetadata(TopicName topicName, long requestId) {
+ Supplier<CompletableFuture<Boolean>> newTopicCheck = newTopicCheck(topicName);
+ AtomicBoolean creationRefused = new AtomicBoolean();
+ AtomicBoolean creationCheckFailed = new AtomicBoolean();
return FutureUtil.supplySafely(() ->
- unsafeGetPartitionedTopicMetadataAsync(getBrokerService().pulsar(), topicName))
+ unsafeGetPartitionedTopicMetadataAsync(getBrokerService().pulsar(), topicName,
+ () -> FutureUtil.supplySafely(newTopicCheck).whenComplete((isAllowed, checkEx) -> {
+ if (checkEx != null) {
+ creationCheckFailed.set(true);
+ } else {
+ creationRefused.set(!isAllowed);
+ }
+ })))
.handle((metadata, ex) -> {
if (ex == null) {
- commandSender.sendPartitionMetadataResponse(metadata.partitions, requestId);
+ if (creationRefused.get()) {
+ // The topic does not exist and this client may not create it, as when the auto topic
+ // creation settings forbid it.
+ writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.TopicNotFound, "",
+ requestId));
+ } else {
+ commandSender.sendPartitionMetadataResponse(metadata.partitions, requestId);
+ }
return null;
}
Throwable actEx = FutureUtil.unwrapCompletionException(ex);
+ if (creationCheckFailed.get()) {
+ // As when checking a topic operation fails.
+ logAuthException(remoteAddress, "partition-metadata", getPrincipal(), Optional.of(topicName),
+ actEx);
+ commandSender.sendPartitionMetadataResponse(ServerError.AuthorizationError,
+ "Exception occurred while checking whether the client may auto-create the topic",
+ requestId);
+ return null;
+ }
if (actEx instanceof PulsarClientException) {
log.warn()
.attr("role", getRole())
@@ -2267,10 +2344,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
return null;
}
- service.isAllowAutoTopicCreationAsync(topicName.toString())
- .thenApply(isAllowed -> forceTopicCreation && isAllowed)
- .thenCompose(createTopicIfDoesNotExist ->
- service.getTopic(topicName.toString(), createTopicIfDoesNotExist))
+ getTopic(topicName, forceTopicCreation)
.thenCompose(optTopic -> {
if (!optTopic.isPresent()) {
return FutureUtil
@@ -2670,7 +2744,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
.attr("schema", schema == null ? "absent" : "present")
.log("Creating producer");
- service.getOrCreateTopic(topicName.toString()).thenComposeAsync((Topic topic) -> {
+ getTopic(topicName, true).thenApply(Optional::get).thenComposeAsync((Topic topic) -> {
// Check max producer limitation to avoid unnecessary ops wasting resources. For example: the new
// producer reached max producer limitation, but pulsar did schema check first, it would waste CPU
if (((AbstractTopic) topic).isProducersExceeded(producerName)) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/DagWatchSession.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/DagWatchSession.java
index 93ce378a2c82..4008a963f015 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/DagWatchSession.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/DagWatchSession.java
@@ -181,13 +181,14 @@ public class DagWatchSession implements ScalableTopicResources.MetadataPathListe
/**
* A lookup for a {@code topic://...} scalable topic that doesn't exist yet. Auto-create it
- * with a single initial segment — gated by the same broker/namespace auto-topic-creation
- * policy as regular topics ({@link BrokerService#isAllowAutoTopicCreationAsync}) — then
- * return its layout. If the policy disallows it, fail with the same not-found error as
- * before so the client sees no behavioural change when auto-creation is off.
+ * with a single initial segment — gated like regular topics by the broker/namespace
+ * auto-topic-creation policy and the client's authorization
+ * ({@link ServerCnx#isAllowAutoTopicCreationAsync}) — then return its layout. If either
+ * disallows it, fail with the same not-found error as before so the client sees no
+ * behavioural change when auto-creation is off.
*/
private CompletableFuture<ScalableTopicLayoutResponse> maybeAutoCreateAndBuildResponse() {
- return brokerService.isAllowAutoTopicCreationAsync(scalableTopicName)
+ return cnx.isAllowAutoTopicCreationAsync(scalableTopicName)
.thenCompose(allowed -> {
if (!allowed) {
return CompletableFuture.failedFuture(
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
index 263133f9268d..3eeb50ca036a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
@@ -1308,6 +1308,34 @@ public abstract class PulsarWebResource {
}
}
+ /**
+ * Whether a topic that does not exist may be created for this client: the auto topic creation settings allow it
+ * and, when it would be created from nothing, the authorization provider lets this client trigger it. The
+ * provider is not consulted for a topic the broker re-materializes from existing metadata, such as a partition
+ * of an existing partitioned topic, see {@link BrokerService#isAllowAutoTopicCreationAsync}.
+ */
+ protected CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(TopicName topicName) {
+ return pulsar().getBrokerService().isAllowAutoTopicCreationAsync(topicName, newTopicCheck(topicName));
+ }
+
+ /**
+ * Whether the authorization provider lets this client trigger the creation of a new topic. The request values are
+ * read now, as the check may run later on another thread.
+ */
+ protected Supplier<CompletableFuture<Boolean>> newTopicCheck(TopicName topicName) {
+ BrokerService brokerService = pulsar().getBrokerService();
+ if (!pulsar().getConfiguration().isAuthenticationEnabled() || !brokerService.isAuthorizationEnabled()) {
+ return () -> CompletableFuture.completedFuture(true);
+ }
+ String originalPrincipal = originalPrincipal();
+ String clientAppId = clientAppId();
+ // As for topic operations, the original principal of a proxied request is checked with its own auth data.
+ AuthenticationDataSource originalAuthData = originalPrincipalAuthData();
+ AuthenticationDataSource authData = clientAuthData();
+ return () -> brokerService.getAuthorizationService().allowTopicAutoCreationAsync(topicName,
+ originalPrincipal, clientAppId, originalAuthData, authData);
+ }
+
public <T> T sync(Supplier<CompletableFuture<T>> supplier) {
try {
return supplier.get().get(config().getMetadataStoreOperationTimeoutSeconds(), SECONDS);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
index 5e871e500780..1c7d7b99763c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
@@ -1934,8 +1934,8 @@ public class ServerCnxTest {
resetChannel();
setChannelConnected();
- CompletableFuture<Topic> delayFuture = new CompletableFuture<>();
- doReturn(delayFuture).when(brokerService).getOrCreateTopic(any(String.class));
+ CompletableFuture<Optional<Topic>> delayFuture = new CompletableFuture<>();
+ doReturn(delayFuture).when(brokerService).getTopicCreatingIfAllowed(any(TopicName.class), any());
// Create producer first time
ByteBuf clientCommand = Commands.newProducer(successTopicName, 1 /* producer id */, 1 /* request id */,
"prod-name", Collections.emptyMap(), false);
@@ -2945,6 +2945,8 @@ public class ServerCnxTest {
doReturn(CompletableFuture.completedFuture(true)).when(authorizationService)
.allowTopicOperationAsync(Mockito.any(),
Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any());
+ doReturn(CompletableFuture.completedFuture(true)).when(authorizationService)
+ .allowTopicAutoCreationAsync(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any());
doReturn(authorizationService).when(brokerService).getAuthorizationService();
svcConfig.setAuthenticationEnabled(true);
svcConfig.setAuthorizationEnabled(true);
@@ -3588,8 +3590,8 @@ public class ServerCnxTest {
resetChannel();
setChannelConnected();
- CompletableFuture<Topic> delayFuture = new CompletableFuture<>();
- doReturn(delayFuture).when(brokerService).getOrCreateTopic(any(String.class));
+ CompletableFuture<Optional<Topic>> delayFuture = new CompletableFuture<>();
+ doReturn(delayFuture).when(brokerService).getTopicCreatingIfAllowed(any(TopicName.class), any());
// Create producer first time
int producerId = 1;
ByteBuf clientCommand = Commands.newProducer(successTopicName, producerId /* producer id */, 1 /* request id */,
@@ -3600,7 +3602,8 @@ public class ServerCnxTest {
channel.writeInbound(closeProducerCmd);
Topic topic = mock(Topic.class);
- doReturn(CompletableFuture.completedFuture(topic)).when(brokerService).getOrCreateTopic(any(String.class));
+ doReturn(CompletableFuture.completedFuture(Optional.of(topic))).when(brokerService)
+ .getTopicCreatingIfAllowed(any(TopicName.class), any());
doReturn(CompletableFuture.completedFuture(false)).when(topic).hasSchema();
clientCommand = Commands.newProducer(successTopicName, producerId /* producer id */, 1 /* request id */,
@@ -4127,12 +4130,11 @@ public class ServerCnxTest {
@DataProvider
public Object[][] partitionMetadataPolicyFailures() {
- return new Object[][]{{true, true}, {true, false}, {false, true}, {false, false}};
+ return new Object[][]{{true}, {false}};
}
@Test(dataProvider = "partitionMetadataPolicyFailures", timeOut = 30000)
- public void testPartitionMetadataPolicyFailure(boolean metadataAutoCreationEnabled, boolean alreadyFailed)
- throws Exception {
+ public void testPartitionMetadataPolicyFailure(boolean alreadyFailed) throws Exception {
resetChannel();
channel.writeInbound(Commands.newConnect("none", "", null));
assertThat(getResponse()).isInstanceOf(CommandConnected.class);
@@ -4148,8 +4150,7 @@ public class ServerCnxTest {
policyFuture.completeExceptionally(failure);
}
- channel.writeInbound(Commands.newPartitionMetadataRequest(successTopicName, requestId,
- metadataAutoCreationEnabled));
+ channel.writeInbound(Commands.newPartitionMetadataRequest(successTopicName, requestId, true));
if (!alreadyFailed) {
assertThat(lookupSemaphore.availablePermits()).isEqualTo(initialPermits - 1);
assertThat(channel.outboundMessages()).isEmpty();
@@ -4169,6 +4170,31 @@ public class ServerCnxTest {
channel.finishAndReleaseAll();
}
+ @Test(timeOut = 30000)
+ public void testPartitionMetadataWithoutAutoCreationSkipsCreationPolicy() throws Exception {
+ resetChannel();
+ channel.writeInbound(Commands.newConnect("none", "", null));
+ assertThat(getResponse()).isInstanceOf(CommandConnected.class);
+
+ TopicName topicName = TopicName.get(successTopicName);
+ // Consulting the auto-creation policy would fail the request with a MetadataError.
+ doReturn(CompletableFuture.failedFuture(new MetadataStoreException("Namespace policy read failed")))
+ .when(brokerService).isAllowAutoTopicCreationAsync(topicName);
+ TopicExistsInfo topicExistsInfo = mock(TopicExistsInfo.class);
+ when(topicExistsInfo.isExists()).thenReturn(true);
+ when(topicExistsInfo.getTopicType()).thenReturn(TopicType.PARTITIONED);
+ when(topicExistsInfo.getPartitions()).thenReturn(3);
+ doReturn(CompletableFuture.completedFuture(topicExistsInfo)).when(namespaceService)
+ .checkTopicExistsAsync(topicName);
+
+ channel.writeInbound(Commands.newPartitionMetadataRequest(successTopicName, 1, false));
+ channel.runPendingTasks();
+ CommandPartitionedTopicMetadataResponse response = (CommandPartitionedTopicMetadataResponse) getResponse();
+ assertThat(response.hasError()).as("error of a lookup that does not ask for auto-creation").isFalse();
+ assertThat(response.getPartitions()).isEqualTo(3);
+ channel.finishAndReleaseAll();
+ }
+
@Test(timeOut = 30000)
public void testPartitionMetadataResponseCleanupFailure() throws Exception {
resetChannel();
@@ -4311,6 +4337,8 @@ public class ServerCnxTest {
CompletableFuture<Boolean> authorizationFuture = new CompletableFuture<>();
doReturn(authorizationFuture).when(authorizationService)
.allowTopicOperationAsync(any(), any(), any(), any(), any(), any());
+ doReturn(CompletableFuture.completedFuture(true)).when(authorizationService)
+ .allowTopicAutoCreationAsync(any(), any(), any(), any(), any());
TopicName topicName = TopicName.get(successTopicName);
Policies policies = new Policies();
@@ -4333,7 +4361,7 @@ public class ServerCnxTest {
doAnswer(invocation -> {
lookupStarted.complete(null);
return metadataFuture;
- }).when(brokerService).fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName);
+ }).when(brokerService).fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(eq(topicName), any());
var lookupSemaphore = brokerService.getLookupRequestSemaphore();
int initialPermits = lookupSemaphore.availablePermits();
@@ -4417,7 +4445,7 @@ public class ServerCnxTest {
doAnswer(invocation -> {
lookupStarted.complete(null);
return metadataFuture;
- }).when(brokerService).fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName);
+ }).when(brokerService).fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(eq(topicName), any());
} else {
doReturn(existingTopicFuture).when(namespaceService).checkTopicExistsAsync(topicName);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/DagWatchSessionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/DagWatchSessionTest.java
index 4e0dec4185e2..3968fe1adae8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/DagWatchSessionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/DagWatchSessionTest.java
@@ -111,13 +111,13 @@ public class DagWatchSessionTest {
@Test
public void testStartFailsWhenTopicMetadataMissingAndAutoCreateDisallowed() {
- // topic://... input + no scalable metadata + auto-create disallowed by policy =
+ // topic://... input + no scalable metadata + auto-create disallowed by policy or authorization =
// TopicNotFound. (When auto-create is allowed the topic is created instead — covered
// end-to-end by V5ScalableTopicAutoCreateTest.) Synthetic layouts are only produced
// for persistent://... input (regular topics).
when(resources.getScalableTopicMetadataAsync(TOPIC, true))
.thenReturn(CompletableFuture.completedFuture(Optional.empty()));
- when(brokerService.isAllowAutoTopicCreationAsync(TOPIC))
+ when(cnx.isAllowAutoTopicCreationAsync(TOPIC))
.thenReturn(CompletableFuture.completedFuture(false));
CompletableFuture<ScalableTopicLayoutResponse> future = session.start();
@@ -161,7 +161,7 @@ public class DagWatchSessionTest {
assertTrue(e.getCause().getMessage().contains("not found"), e.getCause().getMessage());
}
// The opt-out short-circuits before policy is consulted and nothing is created.
- verify(brokerService, never()).isAllowAutoTopicCreationAsync(any(TopicName.class));
+ verify(cnx, never()).isAllowAutoTopicCreationAsync(any(TopicName.class));
verify(brokerService, never()).getScalableTopicService();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicAutoCreationAuthorizationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicAutoCreationAuthorizationTest.java
new file mode 100644
index 000000000000..01a5c543110e
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicAutoCreationAuthorizationTest.java
@@ -0,0 +1,373 @@
+/*
+ * 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.client.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import com.google.common.collect.Sets;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.authentication.AuthenticationDataForwarded;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.namespace.TopicExistsInfo;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.ClientAuthentication;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.TestAuthenticationProvider;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.TestAuthorizationProvider;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.client.impl.auth.v5.V5AuthenticationLoader;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.Policies;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.policies.data.TopicOperation;
+import org.apache.pulsar.common.policies.data.TopicType;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+/**
+ * Topic auto-creation goes through {@code AuthorizationProvider#allowTopicAutoCreationAsync}.
+ *
+ * <p>Runs its own broker rather than {@code SharedPulsarBaseTest}, because it needs authentication and a custom
+ * authorization provider.
+ */
+@Test(groups = "broker-api")
+public class TopicAutoCreationAuthorizationTest extends ProducerConsumerBase {
+
+ private static final String CREATOR_ROLE = "creator";
+ private static final String USER_ROLE = "user";
+ private static final String PROXY_ROLE = "proxy";
+ private static final String NAMESPACE = "my-property/my-ns";
+ /** A namespace whose auto topic creation creates partitioned topics. */
+ private static final String PARTITIONED_NAMESPACE = "my-property/partitioned-ns";
+ private static final int DEFAULT_PARTITIONS = 2;
+
+ private PulsarAdmin superAdmin;
+
+ /** Topics whose name contains it make the auto-creation check fail, as if the provider's backend were down. */
+ private static final String PROVIDER_FAILURE = "provider-failure";
+
+ /**
+ * Lets the client roles and the proxy produce, consume and look up, but only {@link #CREATOR_ROLE} and the proxy
+ * trigger topic auto-creation.
+ */
+ public static class AutoCreationAuthorizationProvider extends TestAuthorizationProvider {
+
+ private static final Set<String> CLIENT_ROLES = Set.of(CREATOR_ROLE, USER_ROLE, PROXY_ROLE);
+
+ /** The auth data each role was last checked with by {@link #allowTopicAutoCreationAsync}. */
+ static final Map<String, AuthenticationDataSource> AUTO_CREATION_AUTH_DATA = new ConcurrentHashMap<>();
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topic, String role,
+ TopicOperation operation,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(CLIENT_ROLES.contains(role));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicAutoCreationAsync(TopicName topic, String role,
+ AuthenticationDataSource authData) {
+ if (authData != null) {
+ AUTO_CREATION_AUTH_DATA.put(role, authData);
+ }
+ if (topic.toString().contains(PROVIDER_FAILURE)) {
+ return CompletableFuture.failedFuture(new IllegalStateException("Authorization backend is down"));
+ }
+ return CompletableFuture.completedFuture(CREATOR_ROLE.equals(role) || PROXY_ROLE.equals(role));
+ }
+ }
+
+ @BeforeMethod(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ conf.setAuthenticationEnabled(true);
+ conf.setAuthorizationEnabled(true);
+ conf.setSuperUserRoles(Set.of("superUser"));
+ conf.setProxyRoles(Set.of(PROXY_ROLE));
+ conf.setAuthenticationProviders(Set.of(TestAuthenticationProvider.class.getName()));
+ conf.setAuthorizationProvider(AutoCreationAuthorizationProvider.class.getName());
+ conf.setAllowAutoTopicCreation(true);
+ conf.setAllowAutoTopicCreationType(TopicType.NON_PARTITIONED);
+ conf.setTopicLevelPoliciesEnabled(false);
+ conf.setClusterName("test");
+ super.init();
+
+ superAdmin = PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
+ .authentication(new ClientAuthentication("superUser")).build();
+ superAdmin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
+ superAdmin.tenants().createTenant("my-property",
+ new TenantInfoImpl(Sets.newHashSet(), Sets.newHashSet("test")));
+ superAdmin.namespaces().createNamespace(NAMESPACE, Sets.newHashSet("test"));
+ // Set at creation: the test authorization provider supports no namespace policy operations.
+ Policies partitionedPolicies = new Policies();
+ partitionedPolicies.replication_clusters = Sets.newHashSet("test");
+ partitionedPolicies.autoTopicCreationOverride = AutoTopicCreationOverride.builder()
+ .allowAutoTopicCreation(true)
+ .topicType(TopicType.PARTITIONED.toString())
+ .defaultNumPartitions(DEFAULT_PARTITIONS)
+ .build();
+ superAdmin.namespaces().createNamespace(PARTITIONED_NAMESPACE, partitionedPolicies);
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (superAdmin != null) {
+ superAdmin.close();
+ superAdmin = null;
+ }
+ super.internalCleanup();
+ }
+
+ @DataProvider
+ public Object[][] lookupTransports() {
+ return new Object[][]{{"binary"}, {"http"}};
+ }
+
+ private String serviceUrl(String lookupTransport) {
+ return "http".equals(lookupTransport) ? brokerUrl.toString() : pulsar.getBrokerServiceUrl();
+ }
+
+ private PulsarClient newClient(String role, String serviceUrl) throws PulsarClientException {
+ return PulsarClient.builder().serviceUrl(serviceUrl)
+ .authentication(new ClientAuthentication(role))
+ .operationTimeout(3, TimeUnit.SECONDS)
+ .build();
+ }
+
+ private boolean topicExists(String topic) throws Exception {
+ TopicExistsInfo info = pulsar.getNamespaceService().checkTopicExistsAsync(TopicName.get(topic)).get();
+ try {
+ return info.isExists();
+ } finally {
+ info.recycle();
+ }
+ }
+
+ @Test(timeOut = 60000)
+ public void testProducerAutoCreationFollowsAuthorization() throws Exception {
+ String topic = "persistent://" + NAMESPACE + "/producer-auto-creation";
+ @Cleanup
+ PulsarClient userClient = newClient(USER_ROLE, pulsar.getBrokerServiceUrl());
+ @Cleanup
+ PulsarClient creatorClient = newClient(CREATOR_ROLE, pulsar.getBrokerServiceUrl());
+
+ assertThatThrownBy(() -> userClient.newProducer().topic(topic).create())
+ .as("a producer of a role that may not create topics")
+ .isInstanceOf(PulsarClientException.TopicDoesNotExistException.class);
+ assertThat(topicExists(topic)).as("topic created by a refused producer").isFalse();
+
+ creatorClient.newProducer().topic(topic).create().close();
+ assertThat(topicExists(topic)).as("topic created by an allowed producer").isTrue();
+
+ // Once the topic exists, a role that may not create topics uses it as before.
+ userClient.newProducer().topic(topic).create().close();
+ }
+
+ @Test(timeOut = 60000)
+ public void testConsumerAutoCreationFollowsAuthorization() throws Exception {
+ String topic = "persistent://" + NAMESPACE + "/consumer-auto-creation";
+ @Cleanup
+ PulsarClient userClient = newClient(USER_ROLE, pulsar.getBrokerServiceUrl());
+ @Cleanup
+ PulsarClient creatorClient = newClient(CREATOR_ROLE, pulsar.getBrokerServiceUrl());
+
+ assertThatThrownBy(() -> userClient.newConsumer().topic(topic).subscriptionName("sub").subscribe())
+ .as("a consumer of a role that may not create topics").isInstanceOf(PulsarClientException.class);
+ assertThat(topicExists(topic)).as("topic created by a refused consumer").isFalse();
+
+ creatorClient.newConsumer().topic(topic).subscriptionName("sub").subscribe().close();
+ assertThat(topicExists(topic)).as("topic created by an allowed consumer").isTrue();
+
+ userClient.newConsumer().topic(topic).subscriptionName("sub").subscribe().close();
+ }
+
+ @Test(timeOut = 60000)
+ public void testHttpLookupAutoCreationFollowsAuthorization() throws Exception {
+ String topic = "persistent://" + NAMESPACE + "/http-lookup-auto-creation";
+ @Cleanup
+ PulsarClient userClient = newClient(USER_ROLE, brokerUrl.toString());
+ @Cleanup
+ PulsarClient creatorClient = newClient(CREATOR_ROLE, brokerUrl.toString());
+
+ assertThatThrownBy(() -> userClient.newProducer().topic(topic).create())
+ .as("a producer looking up over HTTP with a role that may not create topics")
+ .isInstanceOf(PulsarClientException.class);
+ assertThat(topicExists(topic)).as("topic created by a refused HTTP lookup").isFalse();
+
+ creatorClient.newProducer().topic(topic).create().close();
+ assertThat(topicExists(topic)).as("topic created by an allowed HTTP lookup").isTrue();
+ }
+
+ @Test(dataProvider = "lookupTransports", timeOut = 60000)
+ public void testPartitionedAutoCreationFollowsAuthorization(String lookupTransport) throws Exception {
+ String topic = "persistent://" + PARTITIONED_NAMESPACE + "/partitioned-auto-creation-" + lookupTransport;
+ @Cleanup
+ PulsarClient userClient = newClient(USER_ROLE, serviceUrl(lookupTransport));
+ @Cleanup
+ PulsarClient creatorClient = newClient(CREATOR_ROLE, serviceUrl(lookupTransport));
+
+ assertThatThrownBy(() -> userClient.newProducer().topic(topic).create())
+ .as("a producer of a role that may not create partitioned topics")
+ .isInstanceOf(PulsarClientException.class);
+ assertThat(topicExists(topic)).as("partitioned topic created by a refused producer").isFalse();
+
+ creatorClient.newProducer().topic(topic).create().close();
+ assertThat(pulsar.getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topic)).get()
+ .partitions).as("partitions of the auto-created topic").isEqualTo(DEFAULT_PARTITIONS);
+ }
+
+ /**
+ * The partitions of a partitioned topic an admin created are not new topics: a role that may not trigger
+ * auto-creation still gets them created when they are missing, e.g. after inactive-topic deletion removed them.
+ */
+ @Test(dataProvider = "lookupTransports", timeOut = 60000)
+ public void testMissingPartitionsOfExistingPartitionedTopicAreCreated(String lookupTransport) throws Exception {
+ String topic = "persistent://" + NAMESPACE + "/existing-partitioned-" + lookupTransport;
+ // Only the partitioned-topic metadata, without its partitions, as if they had all been deleted.
+ pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .createPartitionedTopicAsync(TopicName.get(topic), new PartitionedTopicMetadata(DEFAULT_PARTITIONS))
+ .get();
+ @Cleanup
+ PulsarClient userClient = newClient(USER_ROLE, serviceUrl(lookupTransport));
+
+ userClient.newProducer().topic(topic).create().close();
+ userClient.newConsumer().topic(topic).subscriptionName("sub").subscribe().close();
+ for (int i = 0; i < DEFAULT_PARTITIONS; i++) {
+ String partition = TopicName.get(topic).getPartition(i).toString();
+ assertThat(topicExists(partition)).as("partition %s created for its existing partitioned topic", i)
+ .isTrue();
+ }
+ }
+
+ @Test(timeOut = 60000)
+ public void testAdminCreateSubscriptionAutoCreationFollowsAuthorization() throws Exception {
+ String topic = "persistent://" + NAMESPACE + "/admin-subscription-auto-creation";
+ @Cleanup
+ PulsarAdmin userAdmin = PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
+ .authentication(new ClientAuthentication(USER_ROLE)).build();
+ @Cleanup
+ PulsarAdmin creatorAdmin = PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
+ .authentication(new ClientAuthentication(CREATOR_ROLE)).build();
+
+ assertThatThrownBy(() -> userAdmin.topics().createSubscription(topic, "sub", MessageId.latest))
+ .as("a subscription created over the admin API by a role that may not create topics")
+ .isInstanceOf(PulsarAdminException.class);
+ assertThat(topicExists(topic)).as("topic created by a refused admin subscription").isFalse();
+
+ creatorAdmin.topics().createSubscription(topic, "sub", MessageId.latest);
+ assertThat(topicExists(topic)).as("topic created by an allowed admin subscription").isTrue();
+ }
+
+ /**
+ * Using a topic that exists does not depend on the auto-creation check, even when it fails.
+ */
+ @Test(dataProvider = "lookupTransports", timeOut = 60000)
+ public void testExistingTopicDoesNotDependOnAutoCreationCheck(String lookupTransport) throws Exception {
+ String topic = "persistent://" + NAMESPACE + "/" + PROVIDER_FAILURE + "-existing-" + lookupTransport;
+ // Created in the metadata, as the admin API does: the test authorization provider lets no admin do it.
+ pulsar.getPulsarResources().getTopicResources().createPersistentTopicAsync(TopicName.get(topic)).get();
+ @Cleanup
+ PulsarClient userClient = newClient(USER_ROLE, serviceUrl(lookupTransport));
+
+ userClient.newProducer().topic(topic).create().close();
+ userClient.newConsumer().topic(topic).subscriptionName("sub").subscribe().close();
+ }
+
+ /**
+ * A proxied HTTP lookup is refused when the original principal may not create the topic, although the proxy
+ * may.
+ */
+ @Test(timeOut = 60000)
+ public void testProxiedHttpLookupAutoCreationFollowsOriginalPrincipal() throws Exception {
+ URI lookupUri = URI.create(brokerUrl.toString().replaceAll("/$", "")
+ + "/lookup/v2/topic/persistent/" + NAMESPACE + "/proxied-http-lookup");
+ HttpClient httpClient = HttpClient.newHttpClient();
+ AutoCreationAuthorizationProvider.AUTO_CREATION_AUTH_DATA.clear();
+
+ assertThat(proxiedLookupStatus(httpClient, lookupUri, USER_ROLE))
+ .as("status of a proxied lookup for a role that may not create topics").isEqualTo(404);
+ // The original principal is checked with its own auth data, not with the proxy's credentials.
+ assertThat(AutoCreationAuthorizationProvider.AUTO_CREATION_AUTH_DATA.get(USER_ROLE))
+ .as("auth data the original principal was checked with")
+ .isInstanceOf(AuthenticationDataForwarded.class);
+ assertThat(AutoCreationAuthorizationProvider.AUTO_CREATION_AUTH_DATA.get(PROXY_ROLE))
+ .as("auth data the proxy was checked with")
+ .isNotNull()
+ .isNotInstanceOf(AuthenticationDataForwarded.class);
+ assertThat(proxiedLookupStatus(httpClient, lookupUri, CREATOR_ROLE))
+ .as("status of a proxied lookup for a role that may create topics").isEqualTo(200);
+ }
+
+ private static int proxiedLookupStatus(HttpClient httpClient, URI lookupUri, String originalPrincipal)
+ throws Exception {
+ HttpRequest request = HttpRequest.newBuilder(lookupUri)
+ .header("user", PROXY_ROLE)
+ .header("X-Original-Principal", originalPrincipal)
+ .GET()
+ .build();
+ return httpClient.send(request, HttpResponse.BodyHandlers.discarding()).statusCode();
+ }
+
+ /**
+ * Auto-creating a scalable topic on lookup ({@code DagWatchSession}) follows the authorization provider too.
+ */
+ @Test(timeOut = 60000)
+ public void testScalableTopicAutoCreationFollowsAuthorization() throws Exception {
+ String topic = "topic://" + NAMESPACE + "/scalable-auto-creation";
+ @Cleanup
+ org.apache.pulsar.client.api.v5.PulsarClient userClient = newV5Client(USER_ROLE);
+ @Cleanup
+ org.apache.pulsar.client.api.v5.PulsarClient creatorClient = newV5Client(CREATOR_ROLE);
+
+ assertThatThrownBy(() -> userClient.newProducer(Schema.string()).topic(topic).create())
+ .as("a producer of a role that may not create scalable topics")
+ .isInstanceOf(org.apache.pulsar.client.api.v5.PulsarClientException.NotFoundException.class);
+ assertThat(scalableTopicExists(topic)).as("scalable topic created by a refused producer").isFalse();
+
+ creatorClient.newProducer(Schema.string()).topic(topic).create().close();
+ assertThat(scalableTopicExists(topic)).as("scalable topic created by an allowed producer").isTrue();
+ }
+
+ private org.apache.pulsar.client.api.v5.PulsarClient newV5Client(String role) throws Exception {
+ return org.apache.pulsar.client.api.v5.PulsarClient.builder()
+ .serviceUrl(pulsar.getBrokerServiceUrl())
+ .authentication(V5AuthenticationLoader.forStartedV4Plugin(new ClientAuthentication(role)))
+ .build();
+ }
+
+ private boolean scalableTopicExists(String topic) throws Exception {
+ return pulsar.getPulsarResources().getScalableTopicResources()
+ .getScalableTopicMetadataAsync(TopicName.get(topic)).get().isPresent();
+ }
+}
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTopicAutoCreationAuthorizationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTopicAutoCreationAuthorizationTest.java
new file mode 100644
index 000000000000..60fdd586a6d3
--- /dev/null
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyTopicAutoCreationAuthorizationTest.java
@@ -0,0 +1,182 @@
+/*
+ * 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.proxy.server;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import com.google.common.collect.Sets;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.authentication.AuthenticationService;
+import org.apache.pulsar.broker.namespace.TopicExistsInfo;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.TestAuthorizationProvider;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.policies.data.TopicOperation;
+import org.apache.pulsar.common.policies.data.TopicType;
+import org.apache.pulsar.proxy.server.ProxyRolesEnforcementTest.BasicAuthentication;
+import org.apache.pulsar.proxy.server.ProxyRolesEnforcementTest.BasicAuthenticationProvider;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Through the proxy, topic auto-creation is subject to the authorization provider for the original principal, not
+ * only for the proxy role.
+ */
+public class ProxyTopicAutoCreationAuthorizationTest extends ProducerConsumerBase {
+
+ private static final String CLUSTER_NAME = "test";
+ private static final String NAMESPACE = "my-property/my-ns";
+ private static final String SUPER_USER_ROLE = "admin";
+ private static final String PROXY_ROLE = "proxy";
+ private static final String CREATOR_ROLE = "creator";
+ private static final String USER_ROLE = "user";
+
+ private Authentication proxyClientAuthentication;
+ private ProxyService proxyService;
+
+ /**
+ * Lets the client roles and the proxy produce, consume and look up, and every role but {@link #USER_ROLE}
+ * trigger topic auto-creation.
+ */
+ public static class AutoCreationAuthorizationProvider extends TestAuthorizationProvider {
+
+ private static final Set<String> ROLES = Set.of(PROXY_ROLE, CREATOR_ROLE, USER_ROLE);
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topic, String role,
+ TopicOperation operation,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(ROLES.contains(role));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicAutoCreationAsync(TopicName topic, String role,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(!USER_ROLE.equals(role));
+ }
+ }
+
+ @BeforeMethod(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ conf.setAuthenticationEnabled(true);
+ conf.setAuthorizationEnabled(true);
+ conf.setAuthenticationProviders(Set.of(BasicAuthenticationProvider.class.getName()));
+ conf.setAuthorizationProvider(AutoCreationAuthorizationProvider.class.getName());
+ conf.setSuperUserRoles(Set.of(SUPER_USER_ROLE));
+ conf.setProxyRoles(Set.of(PROXY_ROLE));
+ conf.setBrokerClientAuthenticationPlugin(BasicAuthentication.class.getName());
+ conf.setBrokerClientAuthenticationParameters("authParam:" + SUPER_USER_ROLE);
+ conf.setAllowAutoTopicCreation(true);
+ conf.setAllowAutoTopicCreationType(TopicType.NON_PARTITIONED);
+ conf.setTopicLevelPoliciesEnabled(false);
+ conf.setClusterName(CLUSTER_NAME);
+ super.init();
+
+ closeAdmin();
+ admin = PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
+ .authentication(BasicAuthentication.class.getName(), "authParam:" + SUPER_USER_ROLE).build();
+ admin.clusters().createCluster(CLUSTER_NAME,
+ ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
+ admin.tenants().createTenant("my-property",
+ new TenantInfoImpl(Sets.newHashSet(), Sets.newHashSet(CLUSTER_NAME)));
+ admin.namespaces().createNamespace(NAMESPACE, Sets.newHashSet(CLUSTER_NAME));
+
+ ProxyConfiguration proxyConfig = new ProxyConfiguration();
+ proxyConfig.setAuthenticationEnabled(true);
+ proxyConfig.setServicePort(Optional.of(0));
+ proxyConfig.setBrokerProxyAllowedTargetPorts("*");
+ proxyConfig.setWebServicePort(Optional.of(0));
+ proxyConfig.setBrokerServiceURL(pulsar.getBrokerServiceUrl());
+ proxyConfig.setBrokerClientAuthenticationPlugin(BasicAuthentication.class.getName());
+ proxyConfig.setBrokerClientAuthenticationParameters("authParam:" + PROXY_ROLE);
+ proxyConfig.setAuthenticationProviders(Set.of(BasicAuthenticationProvider.class.getName()));
+ proxyConfig.setClusterName(CLUSTER_NAME);
+ proxyClientAuthentication = AuthenticationFactory.create(proxyConfig.getBrokerClientAuthenticationPlugin(),
+ proxyConfig.getBrokerClientAuthenticationParameters());
+ proxyClientAuthentication.start();
+ proxyService = new ProxyService(proxyConfig,
+ new AuthenticationService(PulsarConfigurationLoader.convertFrom(proxyConfig)),
+ proxyClientAuthentication);
+ proxyService.start();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (proxyService != null) {
+ proxyService.close();
+ proxyService = null;
+ }
+ if (proxyClientAuthentication != null) {
+ proxyClientAuthentication.close();
+ proxyClientAuthentication = null;
+ }
+ super.internalCleanup();
+ }
+
+ private PulsarClient newProxiedClient(String role) throws PulsarClientException {
+ return PulsarClient.builder().serviceUrl(proxyService.getServiceUrl())
+ .authentication(BasicAuthentication.class.getName(), "authParam:" + role)
+ .operationTimeout(3, TimeUnit.SECONDS)
+ .build();
+ }
+
+ private boolean topicExists(String topic) throws Exception {
+ TopicExistsInfo info = pulsar.getNamespaceService().checkTopicExistsAsync(TopicName.get(topic)).get();
+ try {
+ return info.isExists();
+ } finally {
+ info.recycle();
+ }
+ }
+
+ @Test(timeOut = 60000)
+ public void testAutoCreationFollowsOriginalPrincipal() throws Exception {
+ String topic = "persistent://" + NAMESPACE + "/proxied-auto-creation";
+ @Cleanup
+ PulsarClient userClient = newProxiedClient(USER_ROLE);
+ @Cleanup
+ PulsarClient creatorClient = newProxiedClient(CREATOR_ROLE);
+
+ assertThatThrownBy(() -> userClient.newProducer().topic(topic).create())
+ .as("a proxied producer of a role that may not create topics, although the proxy may")
+ .isInstanceOf(PulsarClientException.TopicDoesNotExistException.class);
+ assertThat(topicExists(topic)).as("topic created by a refused proxied producer").isFalse();
+
+ creatorClient.newProducer().topic(topic).create().close();
+ assertThat(topicExists(topic)).as("topic created by an allowed proxied producer").isTrue();
+
+ userClient.newConsumer().topic(topic).subscriptionName("sub").subscribe().close();
+ }
+}
(pulsar) 01/05: [fix][broker] Fix the broker's WebSocket service ignoring configurationStoreConfigPath (#27015)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0ce235304cd98a97ebb8bf3ffcbcd6ac1c9e2749
Author: Penghui Li <pe...@apache.org>
AuthorDate: Tue Oct 20 09:04:49 2026 -0700
[fix][broker] Fix the broker's WebSocket service ignoring configurationStoreConfigPath (#27015)
(cherry picked from commit de6ca7f7387d339727c4d47b8bcd5988de5b314f)
---
.../org/apache/pulsar/broker/PulsarService.java | 5 +-
.../apache/pulsar/broker/PulsarServiceTest.java | 9 +++
.../apache/pulsar/websocket/WebSocketService.java | 11 ++-
.../WebSocketTlsConfigurationStoreTest.java | 90 ++++++++++++++++++++++
.../src/test/resources/pulsar-websocket.xml | 1 +
5 files changed, 113 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 05550c82f84c..0bfeeb4e9a5e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -1382,8 +1382,9 @@ public class PulsarService implements AutoCloseable, ShutdownService {
ServiceConfiguration config)
throws PulsarServerException, PulsarClientException, MalformedURLException, ServletException {
if (config.isWebSocketServiceEnabled()) {
- // Use local broker address to avoid different IP address when using a VIP for service discovery
- this.webSocketService = new WebSocketService(null, config);
+ // Use local broker address to avoid different IP address when using a VIP for service discovery.
+ // Share the broker's configuration store client, which applies settings like configurationStoreConfigPath.
+ this.webSocketService = new WebSocketService(null, config, pulsarResources);
this.webSocketService.start();
addWebSocketServlet(new WebSocketProducerServlet(webSocketService), attributeMap,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java
index a579705bc142..112f6a48df6b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java
@@ -489,6 +489,15 @@ public class PulsarServiceTest extends MockedPulsarServiceBaseTest {
assertThat(expiryLogged).as("INFO expiry while shuttingDown=%s", shuttingDown).isEqualTo(!shuttingDown);
}
+ @Test
+ public void testWebSocketServiceSharesBrokerConfigurationStore() throws Exception {
+ // Only the broker's own (here mocked) client reaches the configuration store, as with a TLS store set up
+ // through configurationStoreConfigPath; a WebSocket service opening its own client fails the startup.
+ conf.setWebSocketServiceEnabled(true);
+ setup();
+ assertThat(pulsar.getWebSocketService()).as("embedded WebSocket service").isNotNull();
+ }
+
@Test
public void testMetadataSerDesThreads() throws Exception {
final var numSerDesThreads = 5;
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
index 5ba8c8747576..98a304521dd7 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
@@ -88,17 +88,26 @@ public class WebSocketService implements Closeable {
}
public WebSocketService(ClusterData localCluster, ServiceConfiguration config) {
+ this(localCluster, config, null);
+ }
+
+ /**
+ * @param pulsarResources resources to share, such as the broker's, so that no configuration metadata store is
+ * opened by this service; {@code null} to open one in {@link #start()}
+ */
+ public WebSocketService(ClusterData localCluster, ServiceConfiguration config, PulsarResources pulsarResources) {
this.config = config;
this.executor = Executors
.newScheduledThreadPool(config.getWebSocketNumServiceThreads(),
new DefaultThreadFactory("pulsar-websocket"));
this.localCluster = localCluster;
+ this.pulsarResources = pulsarResources;
this.proxyStats = new ProxyStats(this);
}
public void start() throws PulsarServerException, PulsarClientException, MalformedURLException, ServletException {
- if (isNotBlank(config.getConfigurationMetadataStoreUrl())) {
+ if (pulsarResources == null && isNotBlank(config.getConfigurationMetadataStoreUrl())) {
try {
configMetadataStore = createConfigMetadataStore(config.getConfigurationMetadataStoreUrl(),
(int) config.getMetadataStoreSessionTimeoutMillis(),
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/websocket/WebSocketTlsConfigurationStoreTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/websocket/WebSocketTlsConfigurationStoreTest.java
new file mode 100644
index 000000000000..4d7df02e99c2
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/websocket/WebSocketTlsConfigurationStoreTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.tests.integration.websocket;
+
+import static org.apache.pulsar.tests.integration.containers.PulsarContainer.BROKER_HTTP_PORT;
+import static org.apache.pulsar.tests.integration.containers.PulsarContainer.CS_PORT;
+import java.util.Map;
+import org.apache.pulsar.tests.integration.containers.BrokerContainer;
+import org.apache.pulsar.tests.integration.containers.CSContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarClusterSpec;
+import org.testcontainers.images.builder.Transferable;
+import org.testng.annotations.Test;
+
+/**
+ * The broker's WebSocket service when the broker reaches the configuration store over TLS, which only the client
+ * configuration file in {@code configurationStoreConfigPath} sets up. The WebSocket service has to share the
+ * broker's configuration store client: a client of its own would connect in plaintext and fail the broker startup.
+ */
+public class WebSocketTlsConfigurationStoreTest extends WebSocketTestSuite {
+
+ private static final int CS_TLS_PORT = 2281;
+ private static final String CS_CLIENT_CONFIG_PATH = "/pulsar/conf/configuration-store-client.conf";
+
+ @Override
+ protected PulsarClusterSpec.PulsarClusterSpecBuilder beforeSetupCluster(
+ String clusterName,
+ PulsarClusterSpec.PulsarClusterSpecBuilder specBuilder) {
+ specBuilder.numBrokers(1);
+ specBuilder.numProxies(0);
+ specBuilder.brokerEnvs(Map.of(
+ "webSocketServiceEnabled", "true",
+ "configurationMetadataStoreUrl", CSContainer.NAME + ":" + CS_TLS_PORT,
+ // Not in broker.conf, so it is added with the prefix
+ "PULSAR_PREFIX_configurationStoreConfigPath", CS_CLIENT_CONFIG_PATH,
+ // The test certificate is for localhost. Netty 4.2 clients check the hostname by default, which
+ // zookeeper.ssl.hostnameVerification=false doesn't turn off
+ "PULSAR_EXTRA_OPTS", "-Dio.netty.handler.ssl.defaultEndpointVerificationAlgorithm=NONE"));
+ return super.beforeSetupCluster(clusterName, specBuilder);
+ }
+
+ @Override
+ protected void beforeStartCluster() {
+ pulsarCluster.forEachContainer(container -> {
+ if (container instanceof CSContainer) {
+ // A TLS-only port for the broker, and the plaintext one for the cluster metadata initialization
+ container.withCopyToContainer(Transferable.of("""
+ tickTime=2000
+ dataDir=data/global-zookeeper
+ clientPort=%d
+ secureClientPort=%d
+ serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
+ ssl.keyStore.location=/pulsar/certificate-authority/jks/broker.keystore.jks
+ ssl.keyStore.password=111111
+ ssl.clientAuth=none
+ """.formatted(CS_PORT, CS_TLS_PORT)), "/pulsar/conf/global_zookeeper.conf");
+ } else if (container instanceof BrokerContainer) {
+ container.withCopyToContainer(Transferable.of("""
+ zookeeper.client.secure=true
+ zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
+ zookeeper.ssl.trustStore.location=/pulsar/certificate-authority/jks/broker.truststore.jks
+ zookeeper.ssl.trustStore.password=111111
+ # The certificate is for localhost, not for the configuration store's hostname
+ zookeeper.ssl.hostnameVerification=false
+ """), CS_CLIENT_CONFIG_PATH);
+ }
+ });
+ }
+
+ @Test
+ public void testBroker() throws Exception {
+ BrokerContainer broker = pulsarCluster.getAnyBroker();
+ testWebSocket("ws://" + broker.getHost() + ":" + broker.getMappedPort(BROKER_HTTP_PORT));
+ }
+}
diff --git a/tests/integration/src/test/resources/pulsar-websocket.xml b/tests/integration/src/test/resources/pulsar-websocket.xml
index 87bf832d4e40..36f0b8f19744 100644
--- a/tests/integration/src/test/resources/pulsar-websocket.xml
+++ b/tests/integration/src/test/resources/pulsar-websocket.xml
@@ -23,6 +23,7 @@
<test name="messaging-test-suite" preserve-order="true">
<classes>
<class name="org.apache.pulsar.tests.integration.websocket.TestWebSocket" />
+ <class name="org.apache.pulsar.tests.integration.websocket.WebSocketTlsConfigurationStoreTest" />
</classes>
</test>
</suite>
\ No newline at end of file