[fix][broker] Fix various error-prone detected errors mainly in logging and String.format parameters (apache#25059)

(cherry picked from commit 39ea19e)
(cherry picked from commit 6917645)

Author: lhotari

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java

@@ -1163,7 +1163,7 @@ public void deleteSubscriptionDispatchRate(@Suspended AsyncResponse asyncRespons
         internalDeleteSubscriptionDispatchRateAsync()
                 .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
                 .exceptionally(ex -> {
-                    log.error("Failed to delete the subscription dispatchRate for cluster on namespace {}",
+                    log.error("[{}] Failed to delete the subscription dispatchRate for cluster on namespace {}",
                             clientAppId(), namespaceName, ex);
                     resumeAsyncResponseExceptionally(asyncResponse, ex);
                     return null;

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/SplitScheduler.java

@@ -116,7 +116,7 @@ public void execute() {
         synchronized (bundleSplitStrategy) {
             final Set<SplitDecision> decisions = bundleSplitStrategy.findBundlesToSplit(context, pulsar);
             if (debugMode) {
-                log.info("Split Decisions:", decisions);
+                log.info("Split Decisions: {}", decisions);
             }
             if (!decisions.isEmpty()) {
                 // currently following the unloading timeout

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/DefaultNamespaceBundleSplitStrategyImpl.java

@@ -244,7 +244,7 @@ public Set<SplitDecision> findBundlesToSplit(LoadManagerContext context, PulsarS
                         .get(conf.getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
             } catch (Throwable e) {
                 counter.update(Failure, Unknown);
-                log.warn(String.format(CANNOT_SPLIT_BUNDLE_MSG + " Failed to get split boundaries.", bundle, e));
+                log.warn(String.format(CANNOT_SPLIT_BUNDLE_MSG + " Failed to get split boundaries.", bundle), e);
                 continue;
             }
             if (splitBoundary == null) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarMetadataEventSynchronizer.java

@@ -228,11 +228,13 @@ private void startConsumer() {
                 log.info("successfully created consumer {}", topicName);
             } else {
                 State stateTransient = state;
-                log.info("[{}] Closing the new consumer because the synchronizer state is {}", stateTransient);
+                log.info("[{}] Closing the new consumer because the synchronizer state is {}", topicName,
+                        stateTransient);
                 CompletableFuture closeConsumer = new CompletableFuture<>();
                 closeResource(() -> consumer.closeAsync(), closeConsumer);
                 closeConsumer.thenRun(() -> {
-                    log.info("[{}] Closed the new consumer because the synchronizer state is {}", stateTransient);
+                    log.info("[{}] Closed the new consumer because the synchronizer state is {}", topicName,
+                            stateTransient);
                 });
             }
         }).exceptionally(ex -> {
@@ -317,7 +319,7 @@ private void closeResource(final Supplier<CompletableFuture<Void>> asyncCloseabl
             }
             // Retry.
             long waitTimeMs = backOff.next();
-            log.warn("[{}] Exception: '{}' occurred while trying to close the %s. Retrying again in {} s.",
+            log.warn("[{}] Exception: '{}' occurred while trying to close the {}. Retrying again in {} s.",
                     topicName, ex.getMessage(), asyncCloseable.getClass().getSimpleName(), waitTimeMs / 1000.0, ex);
             brokerService.executor().schedule(() -> closeResource(asyncCloseable, future), waitTimeMs,
                     TimeUnit.MILLISECONDS);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java

@@ -64,7 +64,7 @@ protected CompletableFuture<Void> prepareCreateProducer() {
                     if (metadata.partitions == 0) {
                         topicCheckFuture.complete(null);
                     } else {
-                        String errorMsg = String.format("{} Can not create the replicator due to the partitions in the"
+                        String errorMsg = String.format("%s Can not create the replicator due to the partitions in the"
                                         + " remote cluster is not 0, but is %s",
                                 replicatorId, metadata.partitions);
                         log.error(errorMsg);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -2918,13 +2918,12 @@ public CompletableFuture<? extends TopicStatsImpl> asyncGetStats(GetStatsOptions
         stats.topicCreationTimeStamp = getTopicCreationTimeStamp();
 
         stats.compaction.reset();
-        mxBean.flatMap(bean -> bean.getCompactionRecordForTopic(topic)).map(compactionRecord -> {
+        mxBean.flatMap(bean -> bean.getCompactionRecordForTopic(topic)).ifPresent(compactionRecord -> {
             stats.compaction.lastCompactionRemovedEventCount = compactionRecord.getLastCompactionRemovedEventCount();
             stats.compaction.lastCompactionSucceedTimestamp = compactionRecord.getLastCompactionSucceedTimestamp();
             stats.compaction.lastCompactionFailedTimestamp = compactionRecord.getLastCompactionFailedTimestamp();
             stats.compaction.lastCompactionDurationTimeInMills =
                     compactionRecord.getLastCompactionDurationTimeInMills();
-            return compactionRecord;
         });
 
         Map<String, CompletableFuture<SubscriptionStatsImpl>> subscriptionFutures = new HashMap<>();
@@ -2972,7 +2971,7 @@ public CompletableFuture<? extends TopicStatsImpl> asyncGetStats(GetStatsOptions
                 }
             }
             if (getStatsOptions.isGetEarliestTimeInBacklog() && stats.backlogSize != 0) {
-                CompletableFuture finalRes = ledger.getEarliestMessagePublishTimeInBacklog()
+                CompletableFuture<TopicStatsImpl> finalRes = ledger.getEarliestMessagePublishTimeInBacklog()
                     .thenApply((earliestTime) -> {
                         stats.earliestMsgPublishTimeInBacklogs = earliestTime;
                         return stats;

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java

@@ -326,7 +326,7 @@ private static void getTopicStats(Topic topic, TopicStats stats, boolean include
 
         compactorMXBean
                 .flatMap(mxBean -> mxBean.getCompactionRecordForTopic(topic.getName()))
-                .map(compactionRecord -> {
+                .ifPresent(compactionRecord -> {
                     stats.compactionRemovedEventCount = compactionRecord.getCompactionRemovedEventCount();
                     stats.compactionSucceedCount = compactionRecord.getCompactionSucceedCount();
                     stats.compactionFailedCount = compactionRecord.getCompactionFailedCount();
@@ -346,7 +346,6 @@ private static void getTopicStats(Topic topic, TopicStats stats, boolean include
                         stats.compactionCompactedEntriesCount = entries;
                         stats.compactionCompactedEntriesSize = size;
                     }
-                    return compactionRecord;
                 });
     }