Fix idea warning (apache#8588)

### Motivation

When open project in idea, it hints there are something to optimize to make clean code

### Modifications

- Move the param annotation to it's origin class
- The generic can be omit
- Unused imports
- incorrect log params

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

@@ -466,7 +466,7 @@ public void operationComplete(List<String> consumers, Stat s) {
                 }
 
                 if (!ManagedLedgerImpl.this.config.isLazyCursorRecovery()) {
-                    log.debug("[{}] Loading cursor {}", name);
+                    log.debug("[{}] Loading cursors", name);
 
                     for (final String cursorName : consumers) {
                         log.info("[{}] Loading cursor {}", name, cursorName);

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/offload/OffloadUtils.java

@@ -37,7 +37,6 @@
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.OffloadContext;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.OffloadDriverMetadata;
 import org.apache.bookkeeper.net.BookieId;
-import org.apache.bookkeeper.net.BookieSocketAddress;
 import org.apache.bookkeeper.proto.DataFormats;
 
 @Slf4j

pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/PulsarSaslServer.java

@@ -38,7 +38,6 @@
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.common.api.AuthData;
-import org.apache.pulsar.common.sasl.KerberosName;
 import org.apache.pulsar.common.sasl.SaslConstants;
 
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/cache/ConfigurationCacheService.java

@@ -41,7 +41,6 @@
 import org.slf4j.LoggerFactory;
 
 import com.fasterxml.jackson.core.type.TypeReference;
-import com.google.common.collect.Maps;
 
 /**
  * ConfigurationCacheService maintains a local in-memory cache of all the configurations and policies stored in

pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java

@@ -349,7 +349,7 @@ private static BundlesData getBundles(int numBundles) {
         Long maxVal = ((long) 1) << 32;
         Long segSize = maxVal / numBundles;
         List<String> partitions = Lists.newArrayList();
-        partitions.add(String.format("0x%08x", 0l));
+        partitions.add(String.format("0x%08x", 0L));
         Long curPartition = segSize;
         for (int i = 0; i < numBundles; i++) {
             if (i != numBundles - 1) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java

@@ -1421,7 +1421,7 @@ protected void internalSetPublishRate(PublishRate maxPublishMessageRate) {
 
     protected void internalRemovePublishRate() {
         validateSuperUserAccess();
-        log.info("[{}] Remove namespace publish-rate {}/{}", clientAppId(), namespaceName);
+        log.info("[{}] Remove namespace publish-rate {}/{}", clientAppId(), namespaceName, topicName);
         Entry<Policies, Stat> policiesNode = null;
         try {
             final String path = path(POLICIES, namespaceName.toString());
@@ -2442,7 +2442,7 @@ public static BundlesData getBundles(int numBundles) {
         Long maxVal = ((long) 1) << 32;
         Long segSize = maxVal / numBundles;
         List<String> partitions = Lists.newArrayList();
-        partitions.add(String.format("0x%08x", 0l));
+        partitions.add(String.format("0x%08x", 0L));
         Long curPartition = segSize;
         for (int i = 0; i < numBundles; i++) {
             if (i != numBundles - 1) {

pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundles.java

@@ -81,7 +81,7 @@ public NamespaceBundles(NamespaceName nsname, long[] partitions, NamespaceBundle
                 lowerBound = upperBound;
             }
         } else {
-            this.partitions = new long[] { 0l };
+            this.partitions = new long[] { 0L };
             bundles.add(fullBundle);
         }
     }

pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdProduce.java

@@ -44,7 +44,6 @@
 import org.apache.pulsar.client.api.Authentication;
 import org.apache.pulsar.client.api.AuthenticationDataProvider;
 import org.apache.pulsar.client.api.ClientBuilder;
-import org.apache.pulsar.client.api.CompressionType;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.ProducerBuilder;
 import org.apache.pulsar.client.api.PulsarClient;

pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java

@@ -32,7 +32,6 @@
 import java.util.ArrayDeque;
 import java.util.HashSet;
 import java.util.Iterator;
-import java.util.LinkedList;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java

@@ -37,7 +37,6 @@
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
-import org.apache.pulsar.common.api.proto.PulsarApi.IntRange;
 import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData;
 import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataTls.java

@@ -19,16 +19,13 @@
 package org.apache.pulsar.client.impl.auth;
 
 import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
 import java.io.InputStream;
 import java.security.KeyManagementException;
 import java.security.PrivateKey;
 import java.security.cert.Certificate;
 import java.security.cert.X509Certificate;
 import java.util.function.Supplier;
 
-import org.apache.commons.compress.utils.IOUtils;
 import org.apache.pulsar.client.api.AuthenticationDataProvider;
 import org.apache.pulsar.common.util.FileModifiedTimeUpdater;
 import org.apache.pulsar.common.util.SecurityUtility;

pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationTls.java

@@ -20,7 +20,6 @@
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
-import java.io.InputStream;
 import java.util.Map;
 import java.util.function.Supplier;
 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/TransactionImpl.java

@@ -21,15 +21,13 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
 
 import com.google.common.collect.Lists;
-import lombok.Data;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.MessageId;

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntime.java

@@ -260,7 +260,7 @@ public class KubernetesRuntime implements Runtime {
                         authConfig,
                         "$" + ENV_SHARD_ID,
                         grpcPort,
-                        -1l,
+                        -1L,
                         logConfigFile,
                         secretsProviderClassName,
                         secretsProviderConfig,

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java

@@ -51,7 +51,6 @@
 import org.apache.pulsar.functions.runtime.kubernetes.KubernetesRuntimeFactory;
 import org.apache.pulsar.functions.runtime.kubernetes.KubernetesRuntimeFactoryConfig;
 import org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactoryConfig;
-import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactory;
 import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactoryConfig;
 
 @Data

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/ValidatorUtils.java

@@ -21,7 +21,6 @@
 
 import lombok.extern.slf4j.Slf4j;
 import net.jodah.typetools.TypeResolver;
-import org.apache.commons.lang3.reflect.ConstructorUtils;
 import org.apache.pulsar.client.api.CryptoKeyReader;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.common.functions.CryptoConfig;
@@ -33,14 +32,11 @@
 import org.apache.pulsar.io.core.Sink;
 import org.apache.pulsar.io.core.Source;
 
-import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
 import java.util.Map;
 
 import static org.apache.commons.lang3.StringUtils.isBlank;
 import static org.apache.commons.lang3.StringUtils.isEmpty;
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
-import static org.apache.pulsar.common.util.Reflections.createInstance;
 
 @Slf4j
 public class ValidatorUtils {

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/functioncache/FunctionCacheEntry.java

@@ -33,9 +33,6 @@
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
-import static org.apache.commons.lang3.StringUtils.isNoneBlank;
-import static org.apache.pulsar.functions.utils.functioncache.FunctionCacheEntry.JAVA_INSTANCE_JAR_PROPERTY;
-
 import org.apache.pulsar.common.nar.NarClassLoader;
 
 /**

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/MembershipManager.java

@@ -28,20 +28,13 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 import lombok.extern.slf4j.Slf4j;
 
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.ConsumerEventListener;
 import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.client.api.SubscriptionType;
-import org.apache.pulsar.client.impl.ConsumerImpl;
 import org.apache.pulsar.common.functions.WorkerInfo;
 import org.apache.pulsar.common.policies.data.ConsumerStats;
 import org.apache.pulsar.common.policies.data.TopicStats;

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java

@@ -213,7 +213,7 @@ private Future<?> scheduleInternal(Runnable runnable, String errMsg) {
                         try {
                             runnable.run();
                         } catch (Throwable th) {
-                            log.error("Encountered error when invoking scheduler", errMsg);
+                            log.error("Encountered error when invoking scheduler [{}]", errMsg);
                             errorNotifier.triggerError(th);
                         }
                     }

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/zookeeper/ZKMetadataStore.java

@@ -36,7 +36,6 @@
 import org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException;
 import org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException;
 import org.apache.pulsar.metadata.api.Stat;
-import org.apache.zookeeper.AsyncCallback.StatCallback;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.Code;

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java

@@ -131,7 +131,7 @@ public ProxyService(ProxyConfiguration proxyConfig,
                 new Semaphore(proxyConfig.getMaxConcurrentLookupRequests(), false));
 
         if (proxyConfig.getProxyLogLevel().isPresent()) {
-            proxyLogLevel = Integer.valueOf(proxyConfig.getProxyLogLevel().get());
+            proxyLogLevel = proxyConfig.getProxyLogLevel().get();
         } else {
             proxyLogLevel = 0;
         }

pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/GlobalZooKeeperCache.java

@@ -40,7 +40,6 @@
  * and invalidated by the ZNode path. For the data cache, ZNode data parsing is done at request time with the given
  * {@link Deserializer} argument.
  *
- * @param <T>
  */
 public class GlobalZooKeeperCache extends ZooKeeperCache implements Closeable {
     private static final Logger LOG = LoggerFactory.getLogger(GlobalZooKeeperCache.class);

pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java

@@ -85,7 +85,7 @@ public class LocalBookkeeperEnsemble {
     public static final int CONNECTION_TIMEOUT = 30000;
 
     int numberOfBookies;
-    private boolean clearOldData = false;
+    private boolean clearOldData;
 
     private static class BasePortManager implements Supplier<Integer> {
 

pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalZooKeeperCache.java

@@ -29,7 +29,6 @@
  * and invalidated by the ZNode path. For the data cache, ZNode data parsing is done at request time with the given
  * {@link Deserializer} argument.
  *
- * @param <T>
  */
 public class LocalZooKeeperCache extends ZooKeeperCache {
 

pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalZooKeeperConnectionService.java

@@ -24,7 +24,6 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.pulsar.zookeeper.ZooKeeperClientFactory.SessionType;
-import org.apache.pulsar.zookeeper.ZooKeeperSessionWatcher.ShutdownService;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.NoNodeException;
@@ -108,9 +107,7 @@ public static void checkAndCreatePersistNode(ZooKeeper zkc, String path)
 
         // check if the node exists
         if (zkc.exists(path, false) == null) {
-            /**
-             * create znode
-             */
+            //create znode
             try {
                 // do create the node
                 zkc.create(path, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZkIsolatedBookieEnsemblePlacementPolicy.java

@@ -59,8 +59,8 @@ public class ZkIsolatedBookieEnsemblePlacementPolicy extends RackawareEnsemblePl
 
     private ZooKeeperCache bookieMappingCache = null;
 
-    private final List<String> primaryIsolationGroups = new ArrayList<String>();
-    private final List<String> secondaryIsolationGroups = new ArrayList<String>();
+    private final List<String> primaryIsolationGroups = new ArrayList<>();
+    private final List<String> secondaryIsolationGroups = new ArrayList<>();
     private final ObjectMapper jsonMapper = ObjectMapperFactory.create();
 
     public ZkIsolatedBookieEnsemblePlacementPolicy() {
@@ -194,7 +194,7 @@ private Set<BookieId> getBlacklistedBookies(int ensembleSize) {
                 // if primary-isolated-bookies are not enough then add consider secondary isolated bookie group as well.
                 if (totalAvailableBookiesInPrimaryGroup < ensembleSize) {
                     LOG.info(
-                            "Not found enough available-bookies from primary isolation group {} , checking secondary group",
+                            "Not found enough available-bookies from primary isolation group [{}] , checking secondary group [{}]",
                             primaryIsolationGroups, secondaryIsolationGroups);
                     for (String group : secondaryIsolationGroups) {
                         Map<String, BookieInfo> bookieGroup = allGroupsBookieMapping.get(group);

pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZkUtils.java

@@ -85,9 +85,13 @@ public static boolean checkNodeAndWaitExpired(ZooKeeper zk,
      * @return the parent path or null if no parent exists
      */
     public static String getParentForPath(final String path) {
-        if (path == null) return null;
+        if (path == null) {
+            return null;
+        }
         final int length = path.length();
-        if (length == 0 || (length == 1 && path.charAt(0) == '/')) return null;
+        if (length == 0 || (length == 1 && path.charAt(0) == '/')) {
+            return null;
+        }
 
         int partStartIndex = 0;
         char lastChar = path.charAt(0);

pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZooKeeperCache.java

@@ -62,9 +62,13 @@
  * and invalidated by the ZNode path. For the data cache, ZNode data parsing is done at request time with the given
  * {@link Deserializer} argument.
  *
- * @param <T>
  */
 public abstract class ZooKeeperCache implements Watcher {
+
+    /**
+     *
+     * @param <T> the type of zookeeper content
+     */
     public static interface Deserializer<T> {
         T deserialize(String key, byte[] content) throws Exception;
     }

pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZooKeeperChildrenCache.java

@@ -21,7 +21,6 @@
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.pulsar.zookeeper.ZooKeeperCache.CacheUpdater;

pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZooKeeperSessionWatcher.java

@@ -180,6 +180,7 @@ public synchronized void run() {
         }
     }
 
+    @Override
     public void close() {
         if (scheduler != null) {
             scheduler.shutdownNow();