[Minor] Fix java code errors reported by lgtm. (apache#6398)

Four kinds of errors are fixed in this PR:

- Array index out of bounds
- Inconsistent equals and hashCode
- Missing format argument
- Reference equality test of boxed types

According to https://lgtm.com/projects/g/apache/pulsar/alerts/?mode=tree&severity=error&id=&lang=java

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

@@ -195,7 +195,7 @@ public void operationComplete(MLDataFormats.ManagedLedgerInfo mlInfo, Stat stat)
 
                     @Override
                     public void operationFailed(ManagedLedgerException.MetaStoreException e) {
-                        log.warn("[{}] Unable to obtain managed ledger metadata - {}", e);
+                        log.warn("[{}] Unable to obtain managed ledger metadata - {}", managedLedgerName, e);
                         mlMetaCounter.countDown();
                     }
                 });

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/CallbackMutex.java

@@ -62,7 +62,7 @@ public void unlock() {
         if (log.isDebugEnabled()) {
             owner = null;
             position = null;
-            log.debug(">>> Lock {} released token={} at {}", this.hashCode(),
+            log.debug(">>> Lock {} released at {}", this.hashCode(),
                     Thread.currentThread().getStackTrace()[2]);
         }
 

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

@@ -360,7 +360,7 @@ private synchronized void deleteDynamicConfigurationOnZk(String configName) {
                 LOG.info("[{}] Deleted Service configuration {}", clientAppId(), configName);
             } else {
                 if (LOG.isDebugEnabled()) {
-                    LOG.debug("[{}] Can't update non-dynamic configuration {}/{}", clientAppId(), configName);
+                    LOG.debug("[{}] Can't update non-dynamic configuration {}", clientAppId(), configName);
                 }
                 throw new RestException(Status.PRECONDITION_FAILED, " Can't update non-dynamic configuration");
             }

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java

@@ -87,7 +87,7 @@ public CompletableFuture<Builder> requestAsync(final WebTarget target) {
             // auth complete, return a new Builder
             authFuture.whenComplete((respHeaders, ex) -> {
                 if (ex != null) {
-                    log.warn("[{}] Failed to perform http request at authn stage: {}",
+                    log.warn("[{}] Failed to perform http request at auth stage: {}", target.getUri(),
                         ex.getMessage());
                     builderFuture.completeExceptionally(new PulsarClientException(ex));
                     return;

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

@@ -1528,7 +1528,7 @@ public CompletableFuture<Void> seekAsync(MessageId messageId) {
             log.error("[{}][{}] Failed to reset subscription: {}", topic, subscription, e.getCause().getMessage());
             seekFuture.completeExceptionally(
                 PulsarClientException.wrap(e.getCause(),
-                    String.format("[%s][%s] Failed to seek the subscription %s of the topic %s to the message %s",
+                    String.format("Failed to seek the subscription %s of the topic %s to the message %s",
                         subscription, topicName.toString(), messageId.toString())));
             return null;
         });
@@ -1756,6 +1756,14 @@ public int hashCode() {
         return Objects.hash(topic, subscription, consumerName);
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof ConsumerImpl)) return false;
+        ConsumerImpl<?> consumer = (ConsumerImpl<?>) o;
+        return consumerId == consumer.consumerId;
+    }
+
     // wrapper for connection methods
     ClientCnx cnx() {
         return this.connectionHandler.cnx();

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

@@ -1163,7 +1163,7 @@ public void run(Timeout timeout) throws Exception {
             }
 
             if (log.isDebugEnabled()) {
-                log.debug("[{}]  run partitionsAutoUpdateTimerTask for multiTopicsConsumer: {}", topic);
+                log.debug("[{}] run partitionsAutoUpdateTimerTask", topic);
             }
 
             // if last auto update not completed yet, do nothing.

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java

@@ -157,7 +157,7 @@ public boolean equals(Object obj) {
                     && max_unacked_messages_per_subscription == other.max_unacked_messages_per_subscription
                     && compaction_threshold == other.compaction_threshold
                     && offload_threshold == other.offload_threshold
-                    && offload_deletion_lag_ms == other.offload_deletion_lag_ms
+                    && Objects.equals(offload_deletion_lag_ms, other.offload_deletion_lag_ms)
                     && schema_auto_update_compatibility_strategy == other.schema_auto_update_compatibility_strategy
                     && schema_validation_enforced == other.schema_validation_enforced
                     && schema_compatibility_strategy == other.schema_compatibility_strategy

pulsar-common/src/main/java/org/apache/pulsar/common/sasl/KerberosName.java

@@ -271,7 +271,7 @@ static String replaceParameters(String format,
                 if (paramNum != null) {
                     try {
                         int num = Integer.parseInt(paramNum);
-                        if (num < 0 || num > params.length) {
+                        if (num < 0 || num >= params.length) {
                             throw new BadFormatString("index " + num + " from " + format
                                 + " is outside of the valid range 0 to " + (params.length - 1));
                         }