Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Introduce Log4j 2 #20235

Merged
merged 19 commits into from
Sep 1, 2016
Merged

Introduce Log4j 2 #20235

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
5a8f2d7
Disable logger usage checks
jasontedor Aug 30, 2016
7da0cde
Introduce Log4j 2
jasontedor Aug 9, 2016
abf8a1a
Avoid allocating log parameterized messages
jasontedor Aug 30, 2016
0853fc8
Add missing cast to logging message supplier
jasontedor Aug 30, 2016
9a58fc2
Fix failing evil logger tests
jasontedor Aug 30, 2016
abe3efd
Fix failing max map count check test
jasontedor Aug 30, 2016
1f6a4be
Fix failing evil logging configuration tests
jasontedor Aug 31, 2016
4e69ac0
Add link to open logger usage issue
jasontedor Aug 31, 2016
0fdc5ca
Remove logger getter from DeprecationLogger
jasontedor Aug 31, 2016
1d197ed
Mark ESLoggerUsageTests as awaits fix
jasontedor Aug 31, 2016
21dbc5b
Add empty test to ESLoggerUsageTests
jasontedor Aug 31, 2016
ac8c2e9
Enable console logging for CLI tools
jasontedor Aug 31, 2016
e166459
Merge branch 'master' into log4j2
jasontedor Aug 31, 2016
487ffe8
Remove code references to logging.yml
jasontedor Aug 31, 2016
750033d
Update docs for Log4j 2
jasontedor Aug 31, 2016
07d1a72
Update packaging tests for Log4j 2
jasontedor Aug 31, 2016
76ab02e
Merge branch 'master' into log4j2
jasontedor Aug 31, 2016
c98e790
Merge branch 'master' into log4j2
jasontedor Sep 1, 2016
d9064f4
Fix additional exception logging calls
jasontedor Sep 1, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion ...main/java/org/elasticsearch/action/admin/cluster/health/TransportClusterHealthAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.admin.cluster.health;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.IndicesOptions;
Expand Down Expand Up @@ -106,7 +107,7 @@ public void onNoLongerMaster(String source) {

@Override
public void onFailure(String source, Exception e) {
logger.error(new ParameterizedMessage("unexpected failure during [{}]", source), e);
logger.error((Supplier<?>) () -> new ParameterizedMessage("unexpected failure during [{}]", source), e);
Copy link
Member

@pickypg pickypg Aug 30, 2016 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can't help but wonder if we shouldn't have a static helper function that does this for us?

public static Supplier<ParameterizedMessage> message(String message, Object... params) {
  return () -> new ParameterizedMessage(message, params);
}

EDIT: forgot the () -> in the return.

It's of course another import, but it will probably be better than staring at (Supplier<?>) () -> new ParameterizedMessage(...) in so many places.

Copy link
Member

@pickypg pickypg Aug 30, 2016 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Comment got erased by the last commit. I can't help but wonder if we shouldn't have a static helper function that does this for us?

public static Supplier<ParameterizedMessage> message(String message, Object... params) {
  return () -> new ParameterizedMessage(message, params);
}

It's of course another import, but it will probably be better than staring at (Supplier<?>) () -> new ParameterizedMessage(...) in so many places and it removes both of those imports.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe, but then we would want to at least have a lot of extra overrides to avoid allocating a varargs array.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that would be a lot cleaner than what it is here. It can probably be a secondary PR that happens in smaller waves though.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The problem is that these overrides would then introduce boxing on any primitive arguments, and that is something that we want to avoid as well. I do not think it's feasible to do this.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But that's not any different than the current non-exception logging? log4j just creates a ton of overloads with Object parameters too to avoid the params, right?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But that's not any different than the current non-exception logging?

The whole point of going the org.apache.log4j.util.Supplier<?> route was to avoid creating a new instance of ParameterizedMessage if the log message is not going to be emitted. I don't think we want to go through this trouble to immediately turn around and trigger allocations from boxing and varargs (yes, the latter can be avoided with a lot of overrides but the former can not without adding silly overrides for every possible combination).

log4j just creates a ton of overloads with Object parameters too to avoid the params, right?

Yes.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Which boxing are you referring to? I'm not seeing what would be boxed here (that would not have already been boxed with the old code)?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@rjernst The proposal from @pickypg is for a method

public static Supplier<ParameterizedMessage> message(String message, Object... params) {
  return () -> new ParameterizedMessage(message, params);
}

This will cause an allocation of an Object[] and all primitive arguments will be boxed to Objects. Adding overrides like

public static Supplier<ParameterizedMessage> message(String message, Object p1, Object p2) {
  return () -> new ParameterizedMessage(message, params);
}

removes the Object[] allocation, but still any primitive arguments will be boxed to Objects.

These allocations/boxing will occur regardless of whether or not the log level is fine enough to emit the corresponding log message.

With the (Supplier<?>) () -> new ParameterizedMessage("{}", o) pattern (as in this PR) we avoid this boxing unless the log message is going to be emitted.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

but still any primitive arguments will be boxed to Objects.

But this would have been the case with the old methods as well (so nothing would be changing).

Having to do the cast makes logging a pain to use parameterized messages. While creatinga bunch of overloaded helpers may be a pain, maybe it is worth it. Or maybe having this pain will lead us to not depend on logging so much, so it could be good. :) I could go either way.

listener.onFailure(e);
}

Expand Down
3 changes: 2 additions & 1 deletion ...in/java/org/elasticsearch/action/admin/cluster/reroute/TransportClusterRerouteAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
Expand Down Expand Up @@ -104,7 +105,7 @@ public void onAckTimeout() {

@Override
public void onFailure(String source, Exception e) {
logger.debug(new ParameterizedMessage("failed to perform [{}]", source), e);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to perform [{}]", source), e);
super.onFailure(source, e);
}

Expand Down
5 changes: 3 additions & 2 deletions ...org/elasticsearch/action/admin/cluster/settings/TransportClusterUpdateSettingsAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.admin.cluster.settings;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
Expand Down Expand Up @@ -149,7 +150,7 @@ public void onNoLongerMaster(String source) {
@Override
public void onFailure(String source, Exception e) {
//if the reroute fails we only log
logger.debug(new ParameterizedMessage("failed to perform [{}]", source), e);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to perform [{}]", source), e);
listener.onFailure(new ElasticsearchException("reroute after update settings failed", e));
}

Expand All @@ -167,7 +168,7 @@ public ClusterState execute(final ClusterState currentState) {

@Override
public void onFailure(String source, Exception e) {
logger.debug(new ParameterizedMessage("failed to perform [{}]", source), e);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to perform [{}]", source), e);
super.onFailure(source, e);
}

Expand Down
3 changes: 2 additions & 1 deletion ...src/main/java/org/elasticsearch/action/admin/indices/close/TransportCloseIndexAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.admin.indices.close;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.DestructiveOperations;
Expand Down Expand Up @@ -109,7 +110,7 @@ public void onResponse(ClusterStateUpdateResponse response) {

@Override
public void onFailure(Exception t) {
logger.debug(new ParameterizedMessage("failed to close indices [{}]", (Object) concreteIndices), t);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to close indices [{}]", (Object) concreteIndices), t);
listener.onFailure(t);
}
});
Expand Down
3 changes: 2 additions & 1 deletion ...c/main/java/org/elasticsearch/action/admin/indices/delete/TransportDeleteIndexAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.admin.indices.delete;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.DestructiveOperations;
Expand Down Expand Up @@ -101,7 +102,7 @@ public void onResponse(ClusterStateUpdateResponse response) {

@Override
public void onFailure(Exception t) {
logger.debug(new ParameterizedMessage("failed to delete indices [{}]", concreteIndices), t);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to delete indices [{}]", concreteIndices), t);
listener.onFailure(t);
}
});
Expand Down
5 changes: 3 additions & 2 deletions ...in/java/org/elasticsearch/action/admin/indices/mapping/put/TransportPutMappingAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.admin.indices.mapping.put;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
Expand Down Expand Up @@ -93,12 +94,12 @@ public void onResponse(ClusterStateUpdateResponse response) {

@Override
public void onFailure(Exception t) {
logger.debug(new ParameterizedMessage("failed to put mappings on indices [{}], type [{}]", concreteIndices, request.type()), t);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to put mappings on indices [{}], type [{}]", concreteIndices, request.type()), t);
listener.onFailure(t);
}
});
} catch (IndexNotFoundException ex) {
logger.debug(new ParameterizedMessage("failed to put mappings on indices [{}], type [{}]", request.indices(), request.type()), ex);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to put mappings on indices [{}], type [{}]", request.indices(), request.type()), ex);
throw ex;
}
}
Expand Down
3 changes: 2 additions & 1 deletion core/src/main/java/org/elasticsearch/action/admin/indices/open/TransportOpenIndexAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.admin.indices.open;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.DestructiveOperations;
Expand Down Expand Up @@ -94,7 +95,7 @@ public void onResponse(ClusterStateUpdateResponse response) {

@Override
public void onFailure(Exception t) {
logger.debug(new ParameterizedMessage("failed to open indices [{}]", (Object) concreteIndices), t);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to open indices [{}]", (Object) concreteIndices), t);
listener.onFailure(t);
}
});
Expand Down
3 changes: 2 additions & 1 deletion ...va/org/elasticsearch/action/admin/indices/settings/put/TransportUpdateSettingsAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.admin.indices.settings.put;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
Expand Down Expand Up @@ -93,7 +94,7 @@ public void onResponse(ClusterStateUpdateResponse response) {

@Override
public void onFailure(Exception t) {
logger.debug(new ParameterizedMessage("failed to update settings on indices [{}]", (Object) concreteIndices), t);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to update settings on indices [{}]", (Object) concreteIndices), t);
listener.onFailure(t);
}
});
Expand Down
3 changes: 2 additions & 1 deletion ...lasticsearch/action/admin/indices/template/delete/TransportDeleteIndexTemplateAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.elasticsearch.action.admin.indices.template.delete;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
Expand Down Expand Up @@ -74,7 +75,7 @@ public void onResponse(MetaDataIndexTemplateService.RemoveResponse response) {

@Override
public void onFailure(Exception e) {
logger.debug(new ParameterizedMessage("failed to delete templates [{}]", request.name()), e);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to delete templates [{}]", request.name()), e);
listener.onFailure(e);
}
});
Expand Down
3 changes: 2 additions & 1 deletion .../org/elasticsearch/action/admin/indices/template/put/TransportPutIndexTemplateAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.elasticsearch.action.admin.indices.template.put;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
Expand Down Expand Up @@ -95,7 +96,7 @@ public void onResponse(MetaDataIndexTemplateService.PutResponse response) {

@Override
public void onFailure(Exception e) {
logger.debug(new ParameterizedMessage("failed to put template [{}]", request.name()), e);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to put template [{}]", request.name()), e);
listener.onFailure(e);
}
});
Expand Down
3 changes: 2 additions & 1 deletion ...a/org/elasticsearch/action/admin/indices/upgrade/post/TransportUpgradeSettingsAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.admin.indices.upgrade.post;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
Expand Down Expand Up @@ -80,7 +81,7 @@ public void onResponse(ClusterStateUpdateResponse response) {

@Override
public void onFailure(Exception t) {
logger.debug(new ParameterizedMessage("failed to upgrade minimum compatibility version settings on indices [{}]", request.versions().keySet()), t);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to upgrade minimum compatibility version settings on indices [{}]", request.versions().keySet()), t);
listener.onFailure(t);
}
});
Expand Down
9 changes: 5 additions & 4 deletions core/src/main/java/org/elasticsearch/action/bulk/BulkRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.logging.Loggers;
Expand Down Expand Up @@ -77,12 +78,12 @@ public void execute(BulkRequest bulkRequest, long executionId) {
listener.afterBulk(executionId, bulkRequest, bulkResponse);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.info(new ParameterizedMessage("Bulk request {} has been cancelled.", executionId), e);
logger.info((Supplier<?>) () -> new ParameterizedMessage("Bulk request {} has been cancelled.", executionId), e);
if (!afterCalled) {
listener.afterBulk(executionId, bulkRequest, e);
}
} catch (Exception e) {
logger.warn(new ParameterizedMessage("Failed to execute bulk request {}.", executionId), e);
logger.warn((Supplier<?>) () -> new ParameterizedMessage("Failed to execute bulk request {}.", executionId), e);
if (!afterCalled) {
listener.afterBulk(executionId, bulkRequest, e);
}
Expand Down Expand Up @@ -143,10 +144,10 @@ public void onFailure(Exception e) {
bulkRequestSetupSuccessful = true;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.info(new ParameterizedMessage("Bulk request {} has been cancelled.", executionId), e);
logger.info((Supplier<?>) () -> new ParameterizedMessage("Bulk request {} has been cancelled.", executionId), e);
listener.afterBulk(executionId, bulkRequest, e);
} catch (Exception e) {
logger.warn(new ParameterizedMessage("Failed to execute bulk request {}.", executionId), e);
logger.warn((Supplier<?>) () -> new ParameterizedMessage("Failed to execute bulk request {}.", executionId), e);
listener.afterBulk(executionId, bulkRequest, e);
} finally {
if (!bulkRequestSetupSuccessful && acquired) { // if we fail on client.bulk() release the semaphore
Expand Down
5 changes: 3 additions & 2 deletions core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.bulk;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.action.ActionRequest;
Expand Down Expand Up @@ -184,9 +185,9 @@ private Translog.Location index(IndexMetaData metaData, BulkShardRequest request

private <ReplicationRequestT extends ReplicationRequest<ReplicationRequestT>> void logFailure(Throwable t, String operation, ShardId shardId, ReplicationRequest<ReplicationRequestT> request) {
if (ExceptionsHelper.status(t) == RestStatus.CONFLICT) {
logger.trace(new ParameterizedMessage("{} failed to execute bulk item ({}) {}", shardId, operation, request), t);
logger.trace((Supplier<?>) () -> new ParameterizedMessage("{} failed to execute bulk item ({}) {}", shardId, operation, request), t);
} else {
logger.debug(new ParameterizedMessage("{} failed to execute bulk item ({}) {}", shardId, operation, request), t);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("{} failed to execute bulk item ({}) {}", shardId, operation, request), t);
}
}

Expand Down
3 changes: 2 additions & 1 deletion core/src/main/java/org/elasticsearch/action/get/TransportShardMultiGetAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.get;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.TransportActions;
Expand Down Expand Up @@ -93,7 +94,7 @@ protected MultiGetShardResponse shardOperation(MultiGetShardRequest request, Sha
if (TransportActions.isShardNotAvailableException(e)) {
throw (ElasticsearchException) e;
} else {
logger.debug(new ParameterizedMessage("{} failed to execute multi_get for [{}]/[{}]", shardId, item.type(), item.id()), e);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("{} failed to execute multi_get for [{}]/[{}]", shardId, item.type(), item.id()), e);
response.add(request.locations.get(i), new MultiGetResponse.Failure(request.index(), item.type(), item.id(), e));
}
}
Expand Down
5 changes: 3 additions & 2 deletions core/src/main/java/org/elasticsearch/action/ingest/IngestActionFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.ingest;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.ActionResponse;
Expand Down Expand Up @@ -91,7 +92,7 @@ public <Response extends ActionResponse> void apply(String action, Response resp
void processIndexRequest(Task task, String action, ActionListener listener, ActionFilterChain chain, IndexRequest indexRequest) {

executionService.executeIndexRequest(indexRequest, t -> {
logger.error(new ParameterizedMessage("failed to execute pipeline [{}]", indexRequest.getPipeline()), t);
logger.error((Supplier<?>) () -> new ParameterizedMessage("failed to execute pipeline [{}]", indexRequest.getPipeline()), t);
listener.onFailure(t);
}, success -> {
// TransportIndexAction uses IndexRequest and same action name on the node that receives the request and the node that
Expand All @@ -106,7 +107,7 @@ void processBulkIndexRequest(Task task, BulkRequest original, String action, Act
long ingestStartTimeInNanos = System.nanoTime();
BulkRequestModifier bulkRequestModifier = new BulkRequestModifier(original);
executionService.executeBulkRequest(() -> bulkRequestModifier, (indexRequest, exception) -> {
logger.debug(new ParameterizedMessage("failed to execute pipeline [{}] for document [{}/{}/{}]", indexRequest.getPipeline(), indexRequest.index(), indexRequest.type(), indexRequest.id()), exception);
logger.debug((Supplier<?>) () -> new ParameterizedMessage("failed to execute pipeline [{}] for document [{}/{}/{}]", indexRequest.getPipeline(), indexRequest.index(), indexRequest.type(), indexRequest.id()), exception);
bulkRequestModifier.markCurrentItemAsFailed(exception);
}, (exception) -> {
if (exception != null) {
Expand Down
Loading