Don't auto create topic when create RawReader

apache · Dec 27, 2023 · ebe004a · ebe004a
1 parent 767949c
commit ebe004a
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 11 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -3437,18 +3437,27 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
     public synchronized void triggerCompaction()
             throws PulsarServerException, AlreadyRunningException {
         if (currentCompaction.isDone()) {
+            try {
+                lock.writeLock().lock();
+                if (isFenced || isClosingOrDeleting) {
+                    log.info("[{}] Topic is closing or deleting, skip triggering compaction", topic);
+                    return;
+                }
 
-            if (strategicCompactionMap.containsKey(topic)) {
-                currentCompaction = brokerService.pulsar().getStrategicCompactor()
-                        .compact(topic, strategicCompactionMap.get(topic));
-            } else {
-                currentCompaction = topicCompactionService.compact().thenApply(x -> null);
+                if (strategicCompactionMap.containsKey(topic)) {
+                    currentCompaction = brokerService.pulsar().getStrategicCompactor()
+                            .compact(topic, strategicCompactionMap.get(topic));
+                } else {
+                    currentCompaction = topicCompactionService.compact().thenApply(x -> null);
+                }
+                currentCompaction.whenComplete((ignore, ex) -> {
+                    if (ex != null) {
+                        log.warn("[{}] Compaction failure.", topic, ex);
+                    }
+                });
+            } finally {
+                lock.writeLock().unlock();
             }
-            currentCompaction.whenComplete((ignore, ex) -> {
-               if (ex != null){
-                   log.warn("[{}] Compaction failure.", topic, ex);
-               }
-            });
         } else {
             throw new AlreadyRunningException("Compaction already in progress");
         }

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
@@ -122,7 +122,7 @@ static class RawConsumerImpl extends ConsumerImpl<byte[]> {
                     MessageId.earliest,
                     0 /* startMessageRollbackDurationInSec */,
                     Schema.BYTES, null,
-                    true
+                    false
             );
             incomingRawMessages = new GrowableArrayBlockingQueue<>();
             pendingRawReceives = new ConcurrentLinkedQueue<>();