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

fix: refresh the rollup index as part of the rollup indexer actions #86992

Merged
merged 17 commits into from
Jul 6, 2022
Merged

fix: refresh the rollup index as part of the rollup indexer actions #86992

Show file tree
Hide file tree
Changes from 14 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
7968c70
fix: move refresh just before the match_all query
salvatore-campagna May 20, 2022
665647d
fix: refresh the rollup index when rollup is done
salvatore-campagna May 31, 2022
d3e9dd1
Revert "fix: move refresh just before the match_all query"
salvatore-campagna Jun 1, 2022
6385514
Merge branch 'master' into fix/81983-security-tests-rollup
elasticmachine Jun 1, 2022
f48593c
fix: log refresh failure
salvatore-campagna Jun 8, 2022
ea442d5
use NoOpClient in place of Mockito mock
salvatore-campagna Jun 8, 2022
2688bb1
fix: remove redundant 'final' modifier
salvatore-campagna Jun 8, 2022
8e11d8c
Merge branch 'master' into fix/81983-security-tests-rollup
elasticmachine Jun 8, 2022
a46af31
fix: move 'onFinish' to subclass
salvatore-campagna Jun 9, 2022
f9bbc68
Merge branch 'master' into fix/81983-security-tests-rollup
elasticmachine Jun 9, 2022
691c54a
fix: just mock the search action call
salvatore-campagna Jun 9, 2022
3bd3a94
fix: remove unnecessary client dependency
salvatore-campagna Jul 5, 2022
c1f70f3
fix: call onResponse from inside and change logging level
salvatore-campagna Jul 5, 2022
4e5fc42
Merge branch 'master' into fix/81983-security-tests-rollup
elasticmachine Jul 5, 2022
16bd5b0
fix: move back onResponse call
salvatore-campagna Jul 5, 2022
24ee30a
fix: remove client dependency
salvatore-campagna Jul 5, 2022
ceb5c6b
fix: move listener callback into the refresh listener
salvatore-campagna Jul 5, 2022
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
25 changes: 23 additions & 2 deletions x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/job/RollupJobTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@
import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
import org.elasticsearch.action.bulk.BulkAction;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
Expand Down Expand Up @@ -161,8 +163,27 @@ protected void doSaveState(IndexerState indexerState, Map<String, Object> positi

@Override
protected void onFinish(ActionListener<Void> listener) {
logger.debug("Finished indexing for job [" + job.getConfig().getId() + "]");
listener.onResponse(null);
final RollupJobConfig jobConfig = job.getConfig();
final ActionListener<RefreshResponse> refreshResponseActionListener = new ActionListener<>() {

@Override
public void onResponse(RefreshResponse refreshResponse) {
logger.trace("refreshing rollup index {} successful for job {}", jobConfig.getRollupIndex(), jobConfig.getId());
listener.onResponse(null);
}

@Override
public void onFailure(Exception e) {
logger.warn(
"refreshing rollup index {} failed for job {} with exception {}",
jobConfig.getRollupIndex(),
jobConfig.getId(),
e
);
listener.onResponse(null);
}
};
client.admin().indices().refresh(new RefreshRequest(jobConfig.getRollupIndex()), refreshResponseActionListener);
}

@Override
Expand Down
9 changes: 6 additions & 3 deletions ...n/rollup/src/test/java/org/elasticsearch/xpack/rollup/job/RollupIndexerIndexingTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchResponseSections;
import org.elasticsearch.action.search.ShardSearchFailure;
import org.elasticsearch.client.internal.Client;
import org.elasticsearch.common.Rounding;
import org.elasticsearch.common.time.DateFormatter;
import org.elasticsearch.core.TimeValue;
Expand All @@ -49,6 +50,7 @@
import org.elasticsearch.search.aggregations.bucket.composite.CompositeAggregation;
import org.elasticsearch.search.aggregations.bucket.composite.CompositeAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.test.client.NoOpClient;
import org.elasticsearch.threadpool.TestThreadPool;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.xpack.core.indexing.IndexerState;
Expand Down Expand Up @@ -705,10 +707,12 @@ private void executeTestCase(
IndexSearcher searcher = new IndexSearcher(reader);
String dateHistoField = config.getGroupConfig().getDateHistogram().getField();
final ThreadPool threadPool = new TestThreadPool(getTestName());
final Client client = new NoOpClient(getTestName());

try {
try (dir; reader; client) {
RollupJob job = new RollupJob(config, Collections.emptyMap());
final SyncRollupIndexer action = new SyncRollupIndexer(
client,
threadPool,
job,
searcher,
Expand All @@ -718,8 +722,6 @@ private void executeTestCase(
rollupConsumer.accept(action.triggerAndWaitForCompletion(now));
} finally {
ThreadPool.terminate(threadPool, 30, TimeUnit.SECONDS);
reader.close();
dir.close();
}
}

Expand Down Expand Up @@ -819,6 +821,7 @@ class SyncRollupIndexer extends RollupIndexer {
private Exception exc;

SyncRollupIndexer(
Client client,
Copy link
Contributor

Choose a reason for hiding this comment

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

I do not think the client constructor parameter is used anywhere. Perhaps you should remove it

ThreadPool threadPool,
RollupJob job,
IndexSearcher searcher,
Expand Down
3 changes: 0 additions & 3 deletions ...ugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/job/RollupIndexerStateTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -435,7 +435,6 @@ public void testAbortAfterCompletion() throws Exception {

// Don't use the indexer's latch because we completely change doNextSearch()
final CountDownLatch doNextSearchLatch = new CountDownLatch(1);

try {
DelayedEmptyRollupIndexer indexer = new DelayedEmptyRollupIndexer(threadPool, job, state, null) {
@Override
Expand Down Expand Up @@ -840,7 +839,6 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws

final ThreadPool threadPool = new TestThreadPool(getTestName());
try {

NonEmptyRollupIndexer indexer = new NonEmptyRollupIndexer(
threadPool,
job,
Expand Down Expand Up @@ -897,7 +895,6 @@ public void testSearchShardFailure() throws Exception {

final ThreadPool threadPool = new TestThreadPool(getTestName());
try {

NonEmptyRollupIndexer indexer = new NonEmptyRollupIndexer(
threadPool,
job,
Expand Down
8 changes: 5 additions & 3 deletions ...ck/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/job/RollupJobTaskTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.ShardSearchFailure;
import org.elasticsearch.client.internal.Client;
Expand Down Expand Up @@ -49,6 +50,7 @@
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
Expand Down Expand Up @@ -609,7 +611,7 @@ public void testTriggerWithoutHeaders() throws Exception {

((ActionListener) invocationOnMock.getArguments()[2]).onResponse(r);
return null;
}).when(client).execute(any(), any(), any());
}).when(client).execute(eq(SearchAction.INSTANCE), any(), any());

SchedulerEngine schedulerEngine = mock(SchedulerEngine.class);
TaskId taskId = new TaskId("node", 123);
Expand Down Expand Up @@ -714,7 +716,7 @@ public void testTriggerWithHeaders() throws Exception {

((ActionListener) invocationOnMock.getArguments()[2]).onResponse(r);
return null;
}).when(client).execute(any(), any(), any());
}).when(client).execute(eq(SearchAction.INSTANCE), any(), any());

SchedulerEngine schedulerEngine = mock(SchedulerEngine.class);
TaskId taskId = new TaskId("node", 123);
Expand Down Expand Up @@ -819,7 +821,7 @@ public void testSaveStateChangesIDScheme() throws Exception {

((ActionListener) invocationOnMock.getArguments()[2]).onResponse(r);
return null;
}).when(client).execute(any(), any(), any());
}).when(client).execute(eq(SearchAction.INSTANCE), any(), any());

SchedulerEngine schedulerEngine = mock(SchedulerEngine.class);
RollupJobStatus status = new RollupJobStatus(IndexerState.STOPPED, null);
Expand Down