honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-03-01 16:39:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update the persistent task status in the background.

Browse Source 
Follow up for #3256
This commit is contained in:
Martijn van Groningen 2018-01-12 12:59:27 +01:00
parent bad5135d35
commit 7a4860452a
2 changed files with 15 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
x-pack/plugin/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java
View File

@ -110,8 +110,15 @@ public class ShardFollowTasksExecutor extends PersistentTasksExecutor<ShardFollo
ShardFollowTask.Status newStatus = new ShardFollowTask.Status(); ShardFollowTask.Status newStatus = new ShardFollowTask.Status();
newStatus.setProcessedGlobalCheckpoint(leaderGlobalCheckPoint); newStatus.setProcessedGlobalCheckpoint(leaderGlobalCheckPoint);
task.updatePersistentStatus(newStatus, ActionListener.wrap( task.updatePersistentStatus(newStatus, ActionListener.wrap(
persistentTask -> prepare(task, params, leaderGlobalCheckPoint), task::markAsFailed) persistentTask -> logger.debug("[{}] Successfully updated global checkpoint to {}",
); leaderShard, leaderGlobalCheckPoint),
updateStatusException -> {
logger.error(new ParameterizedMessage("[{}] Failed to update global checkpoint to {}",
leaderShard, leaderGlobalCheckPoint), updateStatusException);
task.markAsFailed(updateStatusException);
}
));
prepare(task, params, leaderGlobalCheckPoint);
} else { } else {
task.markAsFailed(e); task.markAsFailed(e);
} }
@ -171,7 +178,7 @@ public class ShardFollowTasksExecutor extends PersistentTasksExecutor<ShardFollo
} }
void createChucks(long from, long to) { void createChucks(long from, long to) {
LOGGER.debug("[{}] Creating chunks for operation range [{}] to [{}]", leaderShard, from, to); LOGGER.debug("{} Creating chunks for operation range [{}] to [{}]", leaderShard, from, to);
for (long i = from; i < to; i += batchSize) { for (long i = from; i < to; i += batchSize) {
long v2 = i + batchSize < to ? i + batchSize : to; long v2 = i + batchSize < to ? i + batchSize : to;
chunks.add(new long[]{i == from ? i : i + 1, v2}); chunks.add(new long[]{i == from ? i : i + 1, v2});
@ -179,14 +186,14 @@ public class ShardFollowTasksExecutor extends PersistentTasksExecutor<ShardFollo
} }
void start() { void start() {
LOGGER.debug("[{}] Start coordination of [{}] chunks with [{}] concurrent processors", LOGGER.debug("{} Start coordination of [{}] chunks with [{}] concurrent processors",
leaderShard, chunks.size(), concurrentProcessors); leaderShard, chunks.size(), concurrentProcessors);
for (int i = 0; i < concurrentProcessors; i++) { for (int i = 0; i < concurrentProcessors; i++) {
ccrExecutor.execute(new AbstractRunnable() { ccrExecutor.execute(new AbstractRunnable() {
@Override @Override
public void onFailure(Exception e) { public void onFailure(Exception e) {
assert e != null; assert e != null;
LOGGER.error(() -> new ParameterizedMessage("[{}] Failure starting processor", leaderShard), e); LOGGER.error(() -> new ParameterizedMessage("{} Failure starting processor", leaderShard), e);
postProcessChuck(e); postProcessChuck(e);
} }
@ -204,13 +211,13 @@ public class ShardFollowTasksExecutor extends PersistentTasksExecutor<ShardFollo
postProcessChuck(null); postProcessChuck(null);
return; return;
} }
LOGGER.debug("[{}] Processing chunk [{}/{}]", leaderShard, chunk[0], chunk[1]); LOGGER.debug("{} Processing chunk [{}/{}]", leaderShard, chunk[0], chunk[1]);
ChunkProcessor processor = new ChunkProcessor(client, ccrExecutor, leaderShard, followerShard, e -> { ChunkProcessor processor = new ChunkProcessor(client, ccrExecutor, leaderShard, followerShard, e -> {
if (e == null) { if (e == null) {
LOGGER.debug("[{}] Successfully processed chunk [{}/{}]", leaderShard, chunk[0], chunk[1]); LOGGER.debug("{} Successfully processed chunk [{}/{}]", leaderShard, chunk[0], chunk[1]);
processNextChunk(); processNextChunk();
} else { } else {
LOGGER.error(() -> new ParameterizedMessage("[{}] Failure processing chunk [{}/{}]", LOGGER.error(() -> new ParameterizedMessage("{} Failure processing chunk [{}/{}]",
leaderShard, chunk[0], chunk[1]), e); leaderShard, chunk[0], chunk[1]), e);
postProcessChuck(e); postProcessChuck(e);
} }

1
x-pack/plugin/src/test/java/org/elasticsearch/xpack/ccr/action/ChunksCoordinatorTests.java
View File

@ -31,7 +31,6 @@ import static org.hamcrest.CoreMatchers.sameInstance;
import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.is;
import static org.mockito.Matchers.any; import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.same; import static org.mockito.Matchers.same;
import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.doThrow;