[improve][misc] Sync commits from apache into 3.1_ds

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java

@@ -138,8 +138,15 @@ public void initiate() {
             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".

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java

@@ -45,6 +45,8 @@
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.authorization.AuthorizationService;
 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;
@@ -163,6 +165,10 @@ public CompletableFuture<Void> validatePoliciesReadOnlyAccessAsync() {
 
     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);
@@ -198,6 +204,8 @@ private CompletableFuture<Void> tryCreatePartitionAsync(final int partition) {
                     }
                     return null;
                 });
+        pulsar().getBrokerService().getTopicEventsDispatcher()
+                .notifyOnCompletion(result, topicName.getPartition(partition).toString(), TopicEvent.CREATE);
         return result;
     }
 
@@ -599,6 +607,13 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
                         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(() -> {

pulsar-broker/src/test/java/org/apache/pulsar/broker/TopicEventsListenerTest.java

@@ -265,12 +265,23 @@ public void testTopicAutoGC(String topicTypePersistence, String topicTypePartiti
     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-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.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 boolean test(@Nullable Entry entry) {
         }
     }
 
+    @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);
+    }
+
 }