honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix sync flush total shards statistics (#18766)

This commit is contained in:
Yannick Welsch 2016-06-10 13:39:47 +02:00 committed by GitHub
parent 6ea89004cd
commit a2c506acd3
1 changed files with 7 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
core/src/main/java/org/elasticsearch/indices/flush/SyncedFlushService.java
View File

@ -114,11 +114,9 @@ public class SyncedFlushService extends AbstractComponent implements IndexEventL
final ClusterState state = clusterService.state();
final Index[] concreteIndices = indexNameExpressionResolver.concreteIndices(state, indicesOptions, aliasesOrIndices);
final Map<String, List<ShardsSyncedFlushResult>> results = ConcurrentCollections.newConcurrentMap();
int totalNumberOfShards = 0;
int numberOfShards = 0;
for (Index index : concreteIndices) {
final IndexMetaData indexMetaData = state.metaData().getIndexSafe(index);
totalNumberOfShards += indexMetaData.getTotalNumberOfShards();
numberOfShards += indexMetaData.getNumberOfShards();
results.put(index.getName(), Collections.synchronizedList(new ArrayList<>()));
@ -127,7 +125,6 @@ public class SyncedFlushService extends AbstractComponent implements IndexEventL
listener.onResponse(new SyncedFlushResponse(results));
return;
}
final int finalTotalNumberOfShards = totalNumberOfShards;
final CountDown countDown = new CountDown(numberOfShards);
for (final Index concreteIndex : concreteIndices) {
@ -136,7 +133,7 @@ public class SyncedFlushService extends AbstractComponent implements IndexEventL
final int indexNumberOfShards = indexMetaData.getNumberOfShards();
for (int shard = 0; shard < indexNumberOfShards; shard++) {
final ShardId shardId = new ShardId(indexMetaData.getIndex(), shard);
attemptSyncedFlush(shardId, new ActionListener<ShardsSyncedFlushResult>() {
innerAttemptSyncedFlush(shardId, state, new ActionListener<ShardsSyncedFlushResult>() {
@Override
public void onResponse(ShardsSyncedFlushResult syncedFlushResult) {
results.get(index).add(syncedFlushResult);
@ -148,7 +145,8 @@ public class SyncedFlushService extends AbstractComponent implements IndexEventL
@Override
public void onFailure(Throwable e) {
logger.debug("{} unexpected error while executing synced flush", shardId);
results.get(index).add(new ShardsSyncedFlushResult(shardId, finalTotalNumberOfShards, e.getMessage()));
final int totalShards = indexMetaData.getNumberOfReplicas() + 1;
results.get(index).add(new ShardsSyncedFlushResult(shardId, totalShards, e.getMessage()));
if (countDown.countDown()) {
listener.onResponse(new SyncedFlushResponse(results));
}
@ -185,8 +183,11 @@ public class SyncedFlushService extends AbstractComponent implements IndexEventL
* Synced flush is a best effort operation. The sync id may be written on all, some or none of the copies.
**/
public void attemptSyncedFlush(final ShardId shardId, final ActionListener<ShardsSyncedFlushResult> actionListener) {
innerAttemptSyncedFlush(shardId, clusterService.state(), actionListener);
}
private void innerAttemptSyncedFlush(final ShardId shardId, final ClusterState state, final ActionListener<ShardsSyncedFlushResult> actionListener) {
try {
final ClusterState state = clusterService.state();
final IndexShardRoutingTable shardRoutingTable = getShardRoutingTable(shardId, state);
final List<ShardRouting> activeShards = shardRoutingTable.activeShards();
final int totalShards = shardRoutingTable.getSize();