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 2024/12/11 13:40:15 UTC
(pulsar) branch branch-3.0 updated (f16cf67baf3 -> ef57c2ac44f)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from f16cf67baf3 [fix][sec] Bump commons-io version to 2.18.0 (#23684)
new 8a472955c9c [fix][broker] Catch exception for entry payload interceptor processor (#23683)
new 7dcc25554e6 [fix][admin] Listen partitioned topic creation event (#23680)
new ef57c2ac44f [improve][admin] Opt-out of topic-existence check (#23709)
The 3 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:
.../apache/bookkeeper/mledger/impl/OpAddEntry.java | 11 ++++-
.../apache/pulsar/broker/ServiceConfiguration.java | 6 +++
.../apache/pulsar/broker/admin/AdminResource.java | 15 +++++++
.../broker/admin/impl/PersistentTopicsBase.java | 39 +++++++++++++----
.../pulsar/broker/TopicEventsListenerTest.java | 23 +++++++---
.../apache/pulsar/broker/admin/AdminApiTest.java | 20 +++++++++
.../intercept/MangedLedgerInterceptorImplTest.java | 49 ++++++++++++++++++++++
7 files changed, 147 insertions(+), 16 deletions(-)
(pulsar) 02/03: [fix][admin] Listen partitioned topic creation event (#23680)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 7dcc25554e68f94b6b1dbd7aef9d61a4e71a079c
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Sun Dec 8 22:26:51 2024 +0800
[fix][admin] Listen partitioned topic creation event (#23680)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 0a2ffe4743799dc253814d49fa3fd29b933444ac)
---
.../apache/pulsar/broker/admin/AdminResource.java | 15 ++++++++++++++
.../pulsar/broker/TopicEventsListenerTest.java | 23 ++++++++++++++++------
2 files changed, 32 insertions(+), 6 deletions(-)
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 76add468915..1daf8d53bac 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
@@ -44,6 +44,8 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.resources.ClusterResources;
+import org.apache.pulsar.broker.service.TopicEventsListener.EventStage;
+import org.apache.pulsar.broker.service.TopicEventsListener.TopicEvent;
import org.apache.pulsar.broker.service.plugin.InvalidEntryFilterException;
import org.apache.pulsar.broker.web.PulsarWebResource;
import org.apache.pulsar.broker.web.RestException;
@@ -162,6 +164,10 @@ public abstract class AdminResource extends PulsarWebResource {
protected CompletableFuture<Void> tryCreatePartitionsAsync(int numPartitions) {
if (!topicName.isPersistent()) {
+ for (int i = 0; i < numPartitions; i++) {
+ pulsar().getBrokerService().getTopicEventsDispatcher()
+ .notify(topicName.getPartition(i).toString(), TopicEvent.CREATE, EventStage.SUCCESS);
+ }
return CompletableFuture.completedFuture(null);
}
List<CompletableFuture<Void>> futures = new ArrayList<>(numPartitions);
@@ -197,6 +203,8 @@ public abstract class AdminResource extends PulsarWebResource {
}
return null;
});
+ pulsar().getBrokerService().getTopicEventsDispatcher()
+ .notifyOnCompletion(result, topicName.getPartition(partition).toString(), TopicEvent.CREATE);
return result;
}
@@ -598,6 +606,13 @@ public abstract class AdminResource extends PulsarWebResource {
throw new RestException(Status.CONFLICT, "This topic already exists");
}
})
+ .thenRun(() -> {
+ for (int i = 0; i < numPartitions; i++) {
+ pulsar().getBrokerService().getTopicEventsDispatcher()
+ .notify(topicName.getPartition(i).toString(), TopicEvent.CREATE,
+ EventStage.BEFORE);
+ }
+ })
.thenCompose(__ -> provisionPartitionedTopicPath(numPartitions, createLocalTopicOnly, properties))
.thenCompose(__ -> tryCreatePartitionsAsync(numPartitions))
.thenRun(() -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/TopicEventsListenerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/TopicEventsListenerTest.java
index ceb3c1d0d93..152b4aeeeb2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/TopicEventsListenerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/TopicEventsListenerTest.java
@@ -265,12 +265,23 @@ public class TopicEventsListenerTest extends BrokerTestBase {
private void createTopicAndVerifyEvents(String topicDomain, String topicTypePartitioned, String topicName) throws Exception {
final String[] expectedEvents;
if (topicDomain.equalsIgnoreCase("persistent") || topicTypePartitioned.equals("partitioned")) {
- expectedEvents = new String[]{
- "LOAD__BEFORE",
- "CREATE__BEFORE",
- "CREATE__SUCCESS",
- "LOAD__SUCCESS"
- };
+ if (topicTypePartitioned.equals("partitioned")) {
+ expectedEvents = new String[]{
+ "CREATE__BEFORE",
+ "CREATE__SUCCESS",
+ "LOAD__BEFORE",
+ "CREATE__BEFORE",
+ "CREATE__SUCCESS",
+ "LOAD__SUCCESS"
+ };
+ } else {
+ expectedEvents = new String[]{
+ "LOAD__BEFORE",
+ "CREATE__BEFORE",
+ "CREATE__SUCCESS",
+ "LOAD__SUCCESS"
+ };
+ }
} else {
expectedEvents = new String[]{
// Before https://github.com/apache/pulsar/pull/21995, Pulsar will skip create topic if the topic
(pulsar) 03/03: [improve][admin] Opt-out of topic-existence check (#23709)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ef57c2ac44fe9b76ef5aaea308f0f7b2c9d0ab47
Author: Omar Yasin <om...@gmail.com>
AuthorDate: Wed Dec 11 04:57:47 2024 -0800
[improve][admin] Opt-out of topic-existence check (#23709)
Co-authored-by: Ómar Yasin <om...@apple.com>
Co-authored-by: Lari Hotari <lh...@apache.org>
(cherry picked from commit f571aa1e6c247bea54d7a10ed508991fee1ea71b)
---
.../apache/pulsar/broker/ServiceConfiguration.java | 6 ++++
.../broker/admin/impl/PersistentTopicsBase.java | 39 +++++++++++++++++-----
.../apache/pulsar/broker/admin/AdminApiTest.java | 20 +++++++++++
3 files changed, 57 insertions(+), 8 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 9b0827426a2..12aba309a01 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -3255,6 +3255,12 @@ public class ServiceConfiguration implements PulsarConfiguration {
)
private int transactionPendingAckBatchedWriteMaxDelayInMillis = 1;
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ doc = "Opt-out of topic-existence check when setting permissions"
+ )
+ private boolean allowAclChangesOnNonExistentTopics = false;
+
/**** --- KeyStore TLS config variables. --- ****/
@FieldContext(
category = CATEGORY_KEYSTORE_TLS,
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 abd9a37aaf7..62335cebc4a 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
@@ -218,8 +218,16 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Map<String, Set<AuthAction>>> internalGetPermissionsOnTopic() {
// This operation should be reading from zookeeper and it should be allowed without having admin privileges
- return validateAdminAccessForTenantAsync(namespaceName.getTenant())
- .thenCompose(__ -> internalCheckTopicExists(topicName))
+ CompletableFuture<Void> validateAccessForTenantCf =
+ validateAdminAccessForTenantAsync(namespaceName.getTenant());
+
+ var checkIfTopicExists = !pulsar().getConfiguration().isAllowAclChangesOnNonExistentTopics();
+ if (checkIfTopicExists) {
+ validateAccessForTenantCf = validateAccessForTenantCf
+ .thenCompose(__ -> internalCheckTopicExists(topicName));
+ }
+
+ return validateAccessForTenantCf
.thenCompose(__ -> namespaceResources().getPoliciesAsync(namespaceName)
.thenApply(policies -> {
if (!policies.isPresent()) {
@@ -299,9 +307,16 @@ public class PersistentTopicsBase extends AdminResource {
protected void internalGrantPermissionsOnTopic(final AsyncResponse asyncResponse, String role,
Set<AuthAction> actions) {
// This operation should be reading from zookeeper and it should be allowed without having admin privileges
- validateAdminAccessForTenantAsync(namespaceName.getTenant())
- .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
- .thenCompose(__ -> internalCheckTopicExists(topicName))
+ CompletableFuture<Void> validateAccessForTenantCf = validateAdminAccessForTenantAsync(namespaceName.getTenant())
+ .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync());
+
+ var checkIfTopicExists = !pulsar().getConfiguration().isAllowAclChangesOnNonExistentTopics();
+ if (checkIfTopicExists) {
+ validateAccessForTenantCf = validateAccessForTenantCf
+ .thenCompose(__ -> internalCheckTopicExists(topicName));
+ }
+
+ validateAccessForTenantCf
.thenCompose(unused1 -> grantPermissionsAsync(topicName, role, actions))
.thenAccept(unused -> asyncResponse.resume(Response.noContent().build()))
.exceptionally(ex -> {
@@ -348,9 +363,17 @@ public class PersistentTopicsBase extends AdminResource {
protected void internalRevokePermissionsOnTopic(AsyncResponse asyncResponse, String role) {
// This operation should be reading from zookeeper and it should be allowed without having admin privileges
- validateAdminAccessForTenantAsync(namespaceName.getTenant())
- .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
- .thenCompose(__ -> internalCheckTopicExists(topicName))
+ CompletableFuture<Void> validateAccessForTenantCf =
+ validateAdminAccessForTenantAsync(namespaceName.getTenant())
+ .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync());
+
+ var checkIfTopicExists = !pulsar().getConfiguration().isAllowAclChangesOnNonExistentTopics();
+ if (checkIfTopicExists) {
+ validateAccessForTenantCf = validateAccessForTenantCf
+ .thenCompose(__ -> internalCheckTopicExists(topicName));
+ }
+
+ validateAccessForTenantCf
.thenCompose(unused1 -> getPartitionedTopicMetadataAsync(topicName, true, false)
.thenCompose(metadata -> {
int numPartitions = metadata.partitions;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
index 697ce784eed..54a34f002d7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.admin;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
@@ -3623,4 +3624,23 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
assertThrows(NotFoundException.class, () -> admin.topics().grantPermission(topic, subject, Set.of(AuthAction.produce)));
assertThrows(NotFoundException.class, () -> admin.topics().revokePermissions(topic, subject));
}
+
+ @Test
+ @SneakyThrows
+ public void testPermissionsAllowAclChangesOnNonExistentTopics() {
+ pulsar.getConfiguration().setAllowAclChangesOnNonExistentTopics(true);
+ try {
+ String namespace = "prop-xyz/ns1/";
+ final String random = UUID.randomUUID().toString();
+ final String topic = "persistent://" + namespace + random;
+ final String subject = UUID.randomUUID().toString();
+ admin.topics().grantPermission(topic, subject, Set.of(AuthAction.produce));
+ assertThat(admin.topics().getPermissions(topic).get(subject)).containsExactly(AuthAction.produce);
+ admin.topics().revokePermissions(topic, subject);
+ assertThat(admin.topics().getPermissions(topic).get(subject)).isNullOrEmpty();
+ } finally {
+ // reset config
+ pulsar.getConfiguration().setAllowAclChangesOnNonExistentTopics(false);
+ }
+ }
}
(pulsar) 01/03: [fix][broker] Catch exception for entry payload interceptor processor (#23683)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8a472955c9cd60043f14471c8735078c5672ec4f
Author: ran <rg...@streamnative.io>
AuthorDate: Sun Dec 8 18:28:55 2024 +0800
[fix][broker] Catch exception for entry payload interceptor processor (#23683)
(cherry picked from commit 24c337f7a586e30a244739a4100697534fbd9f37)
---
.../apache/bookkeeper/mledger/impl/OpAddEntry.java | 11 ++++-
.../intercept/MangedLedgerInterceptorImplTest.java | 49 ++++++++++++++++++++++
2 files changed, 58 insertions(+), 2 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
index acbb0da5a4e..28bbba42558 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
@@ -138,8 +138,15 @@ public class OpAddEntry implements AddCallback, CloseCallback, Runnable {
lastInitTime = System.nanoTime();
if (ml.getManagedLedgerInterceptor() != null) {
long originalDataLen = data.readableBytes();
- payloadProcessorHandle = ml.getManagedLedgerInterceptor().processPayloadBeforeLedgerWrite(this,
- duplicateBuffer);
+ try {
+ payloadProcessorHandle = ml.getManagedLedgerInterceptor()
+ .processPayloadBeforeLedgerWrite(this, duplicateBuffer);
+ } catch (Exception e) {
+ ReferenceCountUtil.safeRelease(duplicateBuffer);
+ log.error("[{}] Error processing payload before ledger write", ml.getName(), e);
+ this.failed(new ManagedLedgerException.ManagedLedgerInterceptException(e));
+ return;
+ }
if (payloadProcessorHandle != null) {
duplicateBuffer = payloadProcessorHandle.getProcessedPayload();
// If data len of entry changes, correct "dataLength" and "currentLedgerSize".
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java
index 7d164b68147..863ab3b9532 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java
@@ -18,8 +18,12 @@
*/
package org.apache.pulsar.broker.intercept;
+import static org.testng.Assert.fail;
+
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.function.Predicate;
@@ -436,4 +440,49 @@ public class MangedLedgerInterceptorImplTest extends MockedBookKeeperTestCase {
}
}
+ @Test(timeOut = 3000)
+ public void testManagedLedgerPayloadInputProcessorFailure() throws Exception {
+ var config = new ManagedLedgerConfig();
+ final String failureMsg = "failed to process input payload";
+ config.setManagedLedgerInterceptor(new ManagedLedgerInterceptorImpl(
+ Collections.emptySet(), Set.of(new ManagedLedgerPayloadProcessor() {
+ @Override
+ public Processor inputProcessor() {
+ return new Processor() {
+ @Override
+ public ByteBuf process(Object contextObj, ByteBuf inputPayload) {
+ throw new RuntimeException(failureMsg);
+ }
+
+ @Override
+ public void release(ByteBuf processedPayload) {
+ // no-op
+ fail("the release method can't be reached");
+ }
+ };
+ }
+ })));
+
+ var ledger = factory.open("testManagedLedgerPayloadProcessorFailure", config);
+ var countDownLatch = new CountDownLatch(1);
+ var expectedException = new ArrayList<Exception>();
+ ledger.asyncAddEntry("test".getBytes(), 1, 1, new AsyncCallbacks.AddEntryCallback() {
+ @Override
+ public void addComplete(Position position, ByteBuf entryData, Object ctx) {
+ entryData.release();
+ countDownLatch.countDown();
+ }
+
+ @Override
+ public void addFailed(ManagedLedgerException exception, Object ctx) {
+ // expected
+ expectedException.add(exception);
+ countDownLatch.countDown();
+ }
+ }, null);
+ countDownLatch.await();
+ assertEquals(expectedException.size(), 1);
+ assertEquals(expectedException.get(0).getCause().getMessage(), failureMsg);
+ }
+
}