Refactor various ParameterizedMessage usages into String.format (#87603)

ParameterizedMessage is part of log4j api and should not be used
in places where a lambda and String.format would be enough.

This commit is the last batch of refactoring to get rid of ParameterizedMessage
in ES codebase and consists of various, not related usages.

relates #86549

server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -8,10 +8,9 @@
 
 package org.elasticsearch.action.bulk;
 
+import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
-import org.apache.logging.log4j.util.MessageSupplier;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRunnable;
@@ -39,6 +38,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.core.Strings;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.IndexingPressure;
@@ -402,17 +402,23 @@ && isConflictException(executionResult.getFailure().getCause())
         } else {
             if (isFailed) {
                 final Exception failure = executionResult.getFailure().getCause();
-                final MessageSupplier messageSupplier = () -> new ParameterizedMessage(
-                    "{} failed to execute bulk item ({}) {}",
-                    context.getPrimary().shardId(),
-                    opType.getLowercase(),
-                    docWriteRequest
-                );
+                Level level;
                 if (TransportShardBulkAction.isConflictException(failure)) {
-                    logger.trace(messageSupplier, failure);
+                    level = Level.TRACE;
                 } else {
-                    logger.debug(messageSupplier, failure);
+                    level = Level.DEBUG;
                 }
+                logger.log(
+                    level,
+                    () -> Strings.format(
+                        "%s failed to execute bulk item (%s) %s",
+                        context.getPrimary().shardId(),
+                        opType.getLowercase(),
+                        docWriteRequest
+                    ),
+                    failure
+                );
+
             }
             response = executionResult;
         }

server/src/main/java/org/elasticsearch/cluster/NodeConnectionsService.java

@@ -10,7 +10,6 @@
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.GroupedActionListener;
 import org.elasticsearch.cluster.coordination.FollowersChecker;
@@ -40,6 +39,7 @@
 
 import static org.elasticsearch.common.settings.Setting.Property;
 import static org.elasticsearch.common.settings.Setting.positiveTimeSetting;
+import static org.elasticsearch.core.Strings.format;
 
 /**
  * This component is responsible for maintaining connections from this node to all the nodes listed in the cluster state, and for
@@ -276,7 +276,7 @@ public void onFailure(Exception e) {
                         final Level level = currentFailureCount % 6 == 1 ? Level.WARN : Level.DEBUG;
                         logger.log(
                             level,
-                            new ParameterizedMessage("failed to connect to {} (tried [{}] times)", discoveryNode, currentFailureCount),
+                            () -> format("failed to connect to %s (tried [%s] times)", discoveryNode, currentFailureCount),
                             e
                         );
                         setConnectionRef(null);

server/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java

@@ -8,10 +8,8 @@
 
 package org.elasticsearch.cluster.action.shard;
 
-import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
@@ -35,7 +33,6 @@
 import org.elasticsearch.cluster.routing.allocation.FailedShard;
 import org.elasticsearch.cluster.routing.allocation.StaleShard;
 import org.elasticsearch.cluster.service.ClusterService;
-import org.elasticsearch.cluster.service.MasterService;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -70,6 +67,9 @@
 import java.util.Set;
 import java.util.function.Predicate;
 
+import static org.apache.logging.log4j.Level.DEBUG;
+import static org.apache.logging.log4j.Level.ERROR;
+import static org.elasticsearch.cluster.service.MasterService.isPublishFailureException;
 import static org.elasticsearch.core.Strings.format;
 
 public class ShardStateAction {
@@ -544,8 +544,8 @@ public record FailedShardUpdateTask(FailedShardEntry entry, ActionListener<Trans
         @Override
         public void onFailure(Exception e) {
             logger.log(
-                MasterService.isPublishFailureException(e) ? Level.DEBUG : Level.ERROR,
-                () -> new ParameterizedMessage("{} unexpected failure while failing shard [{}]", entry.shardId, entry),
+                isPublishFailureException(e) ? DEBUG : ERROR,
+                () -> format("%s unexpected failure while failing shard [%s]", entry.shardId, entry),
                 e
             );
             listener.onFailure(e);

server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java

@@ -10,7 +10,6 @@
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.util.MessageSupplier;
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
@@ -298,14 +297,15 @@ private ClusterFormationState getClusterFormationState() {
         );
     }
 
-    private void onLeaderFailure(MessageSupplier message, Exception e) {
+    private void onLeaderFailure(Supplier<String> message, Exception e) {
         synchronized (mutex) {
             if (mode != Mode.CANDIDATE) {
                 assert lastKnownLeader.isPresent();
                 if (logger.isDebugEnabled()) {
-                    logger.info(message, e);
+                    // TODO this is a workaround for log4j's Supplier. We should remove this, once using ES logging api
+                    logger.info(() -> message.get(), e);
                 } else {
-                    logger.info(message);
+                    logger.info(() -> message.get());
                 }
             }
             becomeCandidate("onLeaderFailure");

server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java

@@ -10,7 +10,6 @@
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ChannelActionListener;
 import org.elasticsearch.cluster.ClusterStateTaskConfig;
@@ -178,11 +177,7 @@ static class FailedJoinAttempt {
         }
 
         void logNow() {
-            logger.log(
-                getLogLevel(exception),
-                () -> new ParameterizedMessage("failed to join {} with {}", destination, joinRequest),
-                exception
-            );
+            logger.log(getLogLevel(exception), () -> format("failed to join %s with %s", destination, joinRequest), exception);
         }
 
         static Level getLogLevel(TransportException e) {

server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java

@@ -10,8 +10,6 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
-import org.apache.logging.log4j.util.MessageSupplier;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
@@ -43,6 +41,7 @@
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
 
 import static org.elasticsearch.core.Strings.format;
 import static org.elasticsearch.monitor.StatusInfo.Status.UNHEALTHY;
@@ -256,8 +255,8 @@ public void handleException(TransportException exp) {
                         if (exp instanceof ConnectTransportException || exp.getCause() instanceof ConnectTransportException) {
                             logger.debug(() -> "leader [" + leader + "] disconnected during check", exp);
                             leaderFailed(
-                                () -> new ParameterizedMessage(
-                                    "master node [{}] disconnected, restarting discovery [{}]",
+                                () -> format(
+                                    "master node [%s] disconnected, restarting discovery [%s]",
                                     leader.descriptionWithoutAttributes(),
                                     ExceptionsHelper.unwrapCause(exp).getMessage()
                                 ),
@@ -267,8 +266,8 @@ public void handleException(TransportException exp) {
                         } else if (exp.getCause() instanceof NodeHealthCheckFailureException) {
                             logger.debug(() -> "leader [" + leader + "] health check failed", exp);
                             leaderFailed(
-                                () -> new ParameterizedMessage(
-                                    "master node [{}] reported itself as unhealthy [{}], {}",
+                                () -> format(
+                                    "master node [%s] reported itself as unhealthy [%s], %s",
                                     leader.descriptionWithoutAttributes(),
                                     exp.getCause().getMessage(),
                                     RESTARTING_DISCOVERY_TEXT
@@ -299,9 +298,9 @@ public void handleException(TransportException exp) {
                                 exp
                             );
                             leaderFailed(
-                                () -> new ParameterizedMessage(
-                                    "[{}] consecutive checks of the master node [{}] were unsuccessful ([{}] rejected, [{}] timed out), "
-                                        + "{} [last unsuccessful check: {}]",
+                                () -> format(
+                                    "[%s] consecutive checks of the master node [%s] were unsuccessful ([%s] rejected, [%s] timed out), "
+                                        + "%s [last unsuccessful check: %s]",
                                     failureCount,
                                     leader.descriptionWithoutAttributes(),
                                     rejectedCountSinceLastSuccess,
@@ -330,7 +329,7 @@ public void handleException(TransportException exp) {
             );
         }
 
-        void leaderFailed(MessageSupplier messageSupplier, Exception e) {
+        void leaderFailed(Supplier<String> messageSupplier, Exception e) {
             if (isClosed.compareAndSet(false, true)) {
                 transportService.getThreadPool().executor(Names.CLUSTER_COORDINATION).execute(new Runnable() {
                     @Override
@@ -352,10 +351,7 @@ void handleDisconnectedNode(DiscoveryNode discoveryNode) {
             if (discoveryNode.equals(leader)) {
                 logger.debug("leader [{}] disconnected", leader);
                 leaderFailed(
-                    () -> new ParameterizedMessage(
-                        "master node [{}] disconnected, restarting discovery",
-                        leader.descriptionWithoutAttributes()
-                    ),
+                    () -> format("master node [%s] disconnected, restarting discovery", leader.descriptionWithoutAttributes()),
                     new NodeDisconnectedException(discoveryNode, "disconnected")
                 );
             }
@@ -429,6 +425,6 @@ interface LeaderFailureListener {
          * @param messageSupplier The message to log if prior to this failure there was a known master in the cluster.
          * @param exception       An exception that gives more detail of the leader failure.
          */
-        void onLeaderFailure(MessageSupplier messageSupplier, Exception exception);
+        void onLeaderFailure(Supplier<String> messageSupplier, Exception exception);
     }
 }

server/src/main/java/org/elasticsearch/rest/BytesRestResponse.java

@@ -8,15 +8,15 @@
 
 package org.elasticsearch.rest;
 
+import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
-import org.apache.logging.log4j.util.Supplier;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.core.Strings;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
@@ -85,16 +85,16 @@ public BytesRestResponse(RestChannel channel, RestStatus status, Exception e) th
         ToXContent.Params params = paramsFromRequest(channel.request());
         if (params.paramAsBoolean(REST_EXCEPTION_SKIP_STACK_TRACE, REST_EXCEPTION_SKIP_STACK_TRACE_DEFAULT) && e != null) {
             // log exception only if it is not returned in the response
-            Supplier<?> messageSupplier = () -> new ParameterizedMessage(
-                "path: {}, params: {}",
-                channel.request().rawPath(),
-                channel.request().params()
-            );
+            Level level = Level.WARN;
             if (status.getStatus() < 500) {
-                SUPPRESSED_ERROR_LOGGER.debug(messageSupplier, e);
-            } else {
-                SUPPRESSED_ERROR_LOGGER.warn(messageSupplier, e);
+                level = Level.DEBUG;
             }
+
+            SUPPRESSED_ERROR_LOGGER.log(
+                level,
+                () -> Strings.format("path: %s, params: %s", channel.request().rawPath(), channel.request().params()),
+                e
+            );
         }
         this.status = status;
         try (XContentBuilder builder = channel.newErrorBuilder()) {

server/src/main/java/org/elasticsearch/transport/TcpTransport.java

@@ -10,7 +10,6 @@
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
@@ -700,8 +699,8 @@ static void handleException(TcpChannel channel, Exception e, Lifecycle lifecycle
                 } else {
                     logger.log(
                         closeConnectionExceptionLevel,
-                        new ParameterizedMessage(
-                            "close connection exception caught on transport layer [{}], disconnecting from relevant node",
+                        () -> format(
+                            "close connection exception caught on transport layer [%s], disconnecting from relevant node",
                             channel
                         ),
                         e

server/src/main/java/org/elasticsearch/upgrades/MigrationResultsUpdateTask.java

@@ -10,7 +10,6 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
@@ -96,11 +95,11 @@ public void onFailure(Exception clusterStateUpdateException) {
             );
         } else {
             logger.error(
-                new ParameterizedMessage(
-                    "failed to update cluster state after failed migration of feature [{}] on index [{}]",
+                () -> format(
+                    "failed to update cluster state after failed migration of feature [%s] on index [%s]",
                     featureName,
                     status.getFailedIndexName()
-                ).getFormattedMessage(),
+                ),
                 clusterStateUpdateException
             );
         }

server/src/main/java/org/elasticsearch/upgrades/SystemIndexMigrationInfo.java

@@ -10,7 +10,6 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.OriginSettingClient;
@@ -30,6 +29,7 @@
 import java.util.stream.Stream;
 
 import static org.elasticsearch.cluster.metadata.IndexMetadata.State.CLOSE;
+import static org.elasticsearch.core.Strings.format;
 
 /**
  * Holds the data required to migrate a single system index, including metadata from the current index. If necessary, computes the settings
@@ -249,23 +249,23 @@ static SystemIndexMigrationInfo fromTaskState(
         // The first case shouldn't happen, master nodes must have all `SystemIndexPlugins` installed.
         // In the second case, we should just start over.
         if (descriptor == null) {
-            String errorMsg = new ParameterizedMessage(
-                "couldn't find system index descriptor for index [{}] from feature [{}], which likely means this node is missing a plugin",
+            String errorMsg = format(
+                "couldn't find system index descriptor for index [%s] from feature [%s], which likely means this node is missing a plugin",
                 taskState.getCurrentIndex(),
                 taskState.getCurrentFeature()
-            ).toString();
+            );
             logger.warn(errorMsg);
             assert false : errorMsg;
             throw new IllegalStateException(errorMsg);
         }
 
         if (imd == null) {
-            String errorMsg = new ParameterizedMessage(
-                "couldn't find index [{}] from feature [{}] with descriptor pattern [{}]",
+            String errorMsg = format(
+                "couldn't find index [%s] from feature [%s] with descriptor pattern [%s]",
                 taskState.getCurrentIndex(),
                 taskState.getCurrentFeature(),
                 descriptor.getIndexPattern()
-            ).toString();
+            );
             logger.warn(errorMsg);
             assert false : errorMsg;
             throw new IllegalStateException(errorMsg);