Skip to content

Commit eeeb5d4

Browse files
original-brownbearjasontedor
original-brownbear
authored and
jasontedor
committed
INGEST: Create Index Before Pipeline Execute (#32786) (#32975)
* INGEST: Create Index Before Pipeline Execute * Ensures that indices are created before the default pipeline setting is read to correcly handle the case of an index template containing a default pipeline (without the fix the first document does not get the pipeline applied as explained in #32758) * closes #32758
1 parent 45e2a18 commit eeeb5d4

File tree

2 files changed

+109
-41
lines changed

2 files changed

+109
-41
lines changed

‎server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java‎

Lines changed: 44 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -131,37 +131,6 @@ protected final void doExecute(final BulkRequest bulkRequest, final ActionListen
131131

132132
@Override
133133
protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<BulkResponse> listener) {
134-
boolean hasIndexRequestsWithPipelines = false;
135-
ImmutableOpenMap<String, IndexMetaData> indicesMetaData = clusterService.state().getMetaData().indices();
136-
for (DocWriteRequest<?> actionRequest : bulkRequest.requests) {
137-
if (actionRequest instanceof IndexRequest) {
138-
IndexRequest indexRequest = (IndexRequest) actionRequest;
139-
String pipeline = indexRequest.getPipeline();
140-
if (pipeline == null) {
141-
IndexMetaData indexMetaData = indicesMetaData.get(indexRequest.index());
142-
if (indexMetaData == null) {
143-
indexRequest.setPipeline(IngestService.NOOP_PIPELINE_NAME);
144-
} else {
145-
String defaultPipeline = IndexSettings.DEFAULT_PIPELINE.get(indexMetaData.getSettings());
146-
indexRequest.setPipeline(defaultPipeline);
147-
if (IngestService.NOOP_PIPELINE_NAME.equals(defaultPipeline) == false) {
148-
hasIndexRequestsWithPipelines = true;
149-
}
150-
}
151-
} else if (IngestService.NOOP_PIPELINE_NAME.equals(pipeline) == false) {
152-
hasIndexRequestsWithPipelines = true;
153-
}
154-
}
155-
}
156-
if (hasIndexRequestsWithPipelines) {
157-
if (clusterService.localNode().isIngestNode()) {
158-
processBulkIndexIngestRequest(task, bulkRequest, listener);
159-
} else {
160-
ingestForwarder.forwardIngestRequest(BulkAction.INSTANCE, bulkRequest, listener);
161-
}
162-
return;
163-
}
164-
165134
final long startTime = relativeTime();
166135
final AtomicArray<BulkItemResponse> responses = new AtomicArray<>(bulkRequest.requests.size());
167136

@@ -195,15 +164,15 @@ protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<Bulk
195164
}
196165
// Step 3: create all the indices that are missing, if there are any missing. start the bulk after all the creates come back.
197166
if (autoCreateIndices.isEmpty()) {
198-
executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated);
167+
executeIngestAndBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated);
199168
} else {
200169
final AtomicInteger counter = new AtomicInteger(autoCreateIndices.size());
201170
for (String index : autoCreateIndices) {
202171
createIndex(index, bulkRequest.timeout(), new ActionListener<CreateIndexResponse>() {
203172
@Override
204173
public void onResponse(CreateIndexResponse result) {
205174
if (counter.decrementAndGet() == 0) {
206-
executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated);
175+
executeIngestAndBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated);
207176
}
208177
}
209178

@@ -219,7 +188,7 @@ public void onFailure(Exception e) {
219188
}
220189
}
221190
if (counter.decrementAndGet() == 0) {
222-
executeBulk(task, bulkRequest, startTime, ActionListener.wrap(listener::onResponse, inner -> {
191+
executeIngestAndBulk(task, bulkRequest, startTime, ActionListener.wrap(listener::onResponse, inner -> {
223192
inner.addSuppressed(e);
224193
listener.onFailure(inner);
225194
}), responses, indicesThatCannotBeCreated);
@@ -229,7 +198,47 @@ public void onFailure(Exception e) {
229198
}
230199
}
231200
} else {
232-
executeBulk(task, bulkRequest, startTime, listener, responses, emptyMap());
201+
executeIngestAndBulk(task, bulkRequest, startTime, listener, responses, emptyMap());
202+
}
203+
}
204+
205+
private void executeIngestAndBulk(Task task, final BulkRequest bulkRequest, final long startTimeNanos,
206+
final ActionListener<BulkResponse> listener, final AtomicArray<BulkItemResponse> responses,
207+
Map<String, IndexNotFoundException> indicesThatCannotBeCreated) {
208+
boolean hasIndexRequestsWithPipelines = false;
209+
ImmutableOpenMap<String, IndexMetaData> indicesMetaData = clusterService.state().getMetaData().indices();
210+
for (DocWriteRequest<?> actionRequest : bulkRequest.requests) {
211+
if (actionRequest instanceof IndexRequest) {
212+
IndexRequest indexRequest = (IndexRequest) actionRequest;
213+
String pipeline = indexRequest.getPipeline();
214+
if (pipeline == null) {
215+
IndexMetaData indexMetaData = indicesMetaData.get(indexRequest.index());
216+
if (indexMetaData == null) {
217+
indexRequest.setPipeline(IngestService.NOOP_PIPELINE_NAME);
218+
} else {
219+
String defaultPipeline = IndexSettings.DEFAULT_PIPELINE.get(indexMetaData.getSettings());
220+
indexRequest.setPipeline(defaultPipeline);
221+
if (IngestService.NOOP_PIPELINE_NAME.equals(defaultPipeline) == false) {
222+
hasIndexRequestsWithPipelines = true;
223+
}
224+
}
225+
} else if (IngestService.NOOP_PIPELINE_NAME.equals(pipeline) == false) {
226+
hasIndexRequestsWithPipelines = true;
227+
}
228+
}
229+
}
230+
if (hasIndexRequestsWithPipelines) {
231+
try {
232+
if (clusterService.localNode().isIngestNode()) {
233+
processBulkIndexIngestRequest(task, bulkRequest, listener);
234+
} else {
235+
ingestForwarder.forwardIngestRequest(BulkAction.INSTANCE, bulkRequest, listener);
236+
}
237+
} catch (Exception e) {
238+
listener.onFailure(e);
239+
}
240+
} else {
241+
executeBulk(task, bulkRequest, startTimeNanos, listener, responses, indicesThatCannotBeCreated);
233242
}
234243
}
235244

‎server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIngestTests.java‎

Lines changed: 65 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -22,20 +22,25 @@
2222
import org.elasticsearch.Version;
2323
import org.elasticsearch.action.ActionListener;
2424
import org.elasticsearch.action.DocWriteRequest;
25+
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
2526
import org.elasticsearch.action.index.IndexAction;
2627
import org.elasticsearch.action.index.IndexRequest;
2728
import org.elasticsearch.action.index.IndexResponse;
2829
import org.elasticsearch.action.support.ActionFilters;
30+
import org.elasticsearch.action.support.AutoCreateIndex;
2931
import org.elasticsearch.cluster.ClusterChangedEvent;
3032
import org.elasticsearch.cluster.ClusterState;
3133
import org.elasticsearch.cluster.ClusterStateApplier;
3234
import org.elasticsearch.cluster.metadata.IndexMetaData;
35+
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
3336
import org.elasticsearch.cluster.metadata.MetaData;
3437
import org.elasticsearch.cluster.node.DiscoveryNode;
3538
import org.elasticsearch.cluster.node.DiscoveryNodes;
3639
import org.elasticsearch.cluster.service.ClusterService;
3740
import org.elasticsearch.common.collect.ImmutableOpenMap;
41+
import org.elasticsearch.common.settings.ClusterSettings;
3842
import org.elasticsearch.common.settings.Settings;
43+
import org.elasticsearch.common.unit.TimeValue;
3944
import org.elasticsearch.common.util.concurrent.AtomicArray;
4045
import org.elasticsearch.index.IndexNotFoundException;
4146
import org.elasticsearch.index.IndexSettings;
@@ -77,6 +82,9 @@ public class TransportBulkActionIngestTests extends ESTestCase {
7782
*/
7883
private static final String WITH_DEFAULT_PIPELINE = "index_with_default_pipeline";
7984

85+
private static final Settings SETTINGS =
86+
Settings.builder().put(AutoCreateIndex.AUTO_CREATE_INDEX_SETTING.getKey(), true).build();
87+
8088
/** Services needed by bulk action */
8189
TransportService transportService;
8290
ClusterService clusterService;
@@ -112,25 +120,42 @@ public class TransportBulkActionIngestTests extends ESTestCase {
112120
/** A subclass of the real bulk action to allow skipping real bulk indexing, and marking when it would have happened. */
113121
class TestTransportBulkAction extends TransportBulkAction {
114122
boolean isExecuted = false; // set when the "real" bulk execution happens
123+
124+
boolean needToCheck; // pluggable return value for `needToCheck`
125+
126+
boolean indexCreated = true; // If set to false, will be set to true by call to createIndex
127+
115128
TestTransportBulkAction() {
116-
super(Settings.EMPTY, null, transportService, clusterService, ingestService,
117-
null, null, new ActionFilters(Collections.emptySet()), null, null);
129+
super(SETTINGS, null, transportService, clusterService, ingestService,
130+
null, null, new ActionFilters(Collections.emptySet()), null,
131+
new AutoCreateIndex(
132+
SETTINGS, new ClusterSettings(SETTINGS, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
133+
new IndexNameExpressionResolver(SETTINGS)
134+
)
135+
);
118136
}
119137
@Override
120138
protected boolean needToCheck() {
121-
return false;
139+
return needToCheck;
122140
}
123141
@Override
124142
void executeBulk(Task task, final BulkRequest bulkRequest, final long startTimeNanos, final ActionListener<BulkResponse> listener,
125143
final AtomicArray<BulkItemResponse> responses, Map<String, IndexNotFoundException> indicesThatCannotBeCreated) {
144+
assertTrue(indexCreated);
126145
isExecuted = true;
127146
}
147+
148+
@Override
149+
void createIndex(String index, TimeValue timeout, ActionListener<CreateIndexResponse> listener) {
150+
indexCreated = true;
151+
listener.onResponse(null);
152+
}
128153
}
129154

130155
class TestSingleItemBulkWriteAction extends TransportSingleItemBulkWriteAction<IndexRequest, IndexResponse> {
131156

132157
TestSingleItemBulkWriteAction(TestTransportBulkAction bulkAction) {
133-
super(Settings.EMPTY, IndexAction.NAME, TransportBulkActionIngestTests.this.transportService,
158+
super(SETTINGS, IndexAction.NAME, TransportBulkActionIngestTests.this.transportService,
134159
TransportBulkActionIngestTests.this.clusterService,
135160
null, null, null, new ActionFilters(Collections.emptySet()), null,
136161
IndexRequest::new, IndexRequest::new, ThreadPool.Names.INDEX, bulkAction, null);
@@ -162,15 +187,17 @@ public void setupAction() {
162187
when(nodes.getIngestNodes()).thenReturn(ingestNodes);
163188
ClusterState state = mock(ClusterState.class);
164189
when(state.getNodes()).thenReturn(nodes);
165-
when(state.getMetaData()).thenReturn(MetaData.builder().indices(ImmutableOpenMap.<String, IndexMetaData>builder()
190+
MetaData metaData = MetaData.builder().indices(ImmutableOpenMap.<String, IndexMetaData>builder()
166191
.putAll(
167192
Collections.singletonMap(
168193
WITH_DEFAULT_PIPELINE,
169194
IndexMetaData.builder(WITH_DEFAULT_PIPELINE).settings(
170195
settings(Version.CURRENT).put(IndexSettings.DEFAULT_PIPELINE.getKey(), "default_pipeline")
171196
.build()
172197
).numberOfShards(1).numberOfReplicas(1).build()))
173-
.build()).build());
198+
.build()).build();
199+
when(state.getMetaData()).thenReturn(metaData);
200+
when(state.metaData()).thenReturn(metaData);
174201
when(clusterService.state()).thenReturn(state);
175202
doAnswer(invocation -> {
176203
ClusterChangedEvent event = mock(ClusterChangedEvent.class);
@@ -408,4 +435,36 @@ public void testUseDefaultPipeline() throws Exception {
408435
verifyZeroInteractions(transportService);
409436
}
410437

438+
public void testCreateIndexBeforeRunPipeline() throws Exception {
439+
Exception exception = new Exception("fake exception");
440+
IndexRequest indexRequest = new IndexRequest("missing_index", "type", "id");
441+
indexRequest.setPipeline("testpipeline");
442+
indexRequest.source(Collections.emptyMap());
443+
AtomicBoolean responseCalled = new AtomicBoolean(false);
444+
AtomicBoolean failureCalled = new AtomicBoolean(false);
445+
action.needToCheck = true;
446+
action.indexCreated = false;
447+
singleItemBulkWriteAction.execute(null, indexRequest, ActionListener.wrap(
448+
response -> responseCalled.set(true),
449+
e -> {
450+
assertThat(e, sameInstance(exception));
451+
failureCalled.set(true);
452+
}));
453+
454+
// check failure works, and passes through to the listener
455+
assertFalse(action.isExecuted); // haven't executed yet
456+
assertFalse(responseCalled.get());
457+
assertFalse(failureCalled.get());
458+
verify(executionService).executeBulkRequest(bulkDocsItr.capture(), failureHandler.capture(), completionHandler.capture());
459+
completionHandler.getValue().accept(exception);
460+
assertTrue(failureCalled.get());
461+
462+
// now check success
463+
indexRequest.setPipeline(IngestService.NOOP_PIPELINE_NAME); // this is done by the real pipeline execution service when processing
464+
completionHandler.getValue().accept(null);
465+
assertTrue(action.isExecuted);
466+
assertFalse(responseCalled.get()); // listener would only be called by real index action, not our mocked one
467+
verifyZeroInteractions(transportService);
468+
}
469+
411470
}

0 commit comments

Comments
 (0)