Skip to content

Commit 04692db

Browse files
gargmanik13gargmanik13
committed
Enabled Async Shard Batch Fetch by default
Signed-off-by: Manik Garg <gargmanik1317@gmail.com>
1 parent b32fcef commit 04692db

File tree

6 files changed

+49
-45
lines changed

6 files changed

+49
-45
lines changed

‎CHANGELOG.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
1111
- Implement parallel shard refresh behind cluster settings ([#17782](https://github.com/opensearch-project/OpenSearch/pull/17782))
1212
- Bump OpenSearch Core main branch to 3.0.0 ([#18039](https://github.com/opensearch-project/OpenSearch/pull/18039))
1313
- Update API of Message in index to add the timestamp for lag calculation in ingestion polling ([#17977](https://github.com/opensearch-project/OpenSearch/pull/17977/))
14-
14+
- Enabled Async Shard Batch Fetch by default ([#17987](https://github.com/opensearch-project/OpenSearch/pull/17987))
1515

1616
### Changed
1717
- Change the default max header size from 8KB to 16KB. ([#18024](https://github.com/opensearch-project/OpenSearch/pull/18024))

‎server/src/internalClusterTest/java/org/opensearch/indices/recovery/IndexRecoveryIT.java‎

Lines changed: 30 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -1505,42 +1505,40 @@ public void testOngoingRecoveryAndClusterManagerFailOver() throws Exception {
15051505
allowToCompletePhase1Latch.await();
15061506
} catch (InterruptedException e) {
15071507
throw new AssertionError(e);
1508+
} finally {
1509+
blockRecovery.release();
15081510
}
15091511
}
15101512
connection.sendRequest(requestId, action, request, options);
15111513
});
1512-
try {
1513-
String nodeWithReplica = internalCluster().startDataOnlyNode();
1514-
assertAcked(
1515-
client().admin()
1516-
.indices()
1517-
.prepareUpdateSettings(indexName)
1518-
.setSettings(
1519-
Settings.builder()
1520-
.put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1)
1521-
.put("index.routing.allocation.include._name", nodeWithPrimary + "," + nodeWithReplica)
1522-
)
1523-
);
1524-
phase1ReadyBlocked.await();
1525-
internalCluster().restartNode(
1526-
clusterService().state().nodes().getClusterManagerNode().getName(),
1527-
new InternalTestCluster.RestartCallback()
1528-
);
1529-
internalCluster().ensureAtLeastNumDataNodes(3);
1530-
assertAcked(
1531-
client().admin()
1532-
.indices()
1533-
.prepareUpdateSettings(indexName)
1534-
.setSettings(
1535-
Settings.builder()
1536-
.put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 2)
1537-
.putNull("index.routing.allocation.include._name")
1538-
)
1539-
);
1540-
assertFalse(client().admin().cluster().prepareHealth(indexName).setWaitForActiveShards(2).get().isTimedOut());
1541-
} finally {
1542-
allowToCompletePhase1Latch.countDown();
1543-
}
1514+
String nodeWithReplica = internalCluster().startDataOnlyNode();
1515+
assertAcked(
1516+
client().admin()
1517+
.indices()
1518+
.prepareUpdateSettings(indexName)
1519+
.setSettings(
1520+
Settings.builder()
1521+
.put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1)
1522+
.put("index.routing.allocation.include._name", nodeWithPrimary + "," + nodeWithReplica)
1523+
)
1524+
);
1525+
phase1ReadyBlocked.await();
1526+
internalCluster().restartNode(
1527+
clusterService().state().nodes().getClusterManagerNode().getName(),
1528+
new InternalTestCluster.RestartCallback()
1529+
);
1530+
internalCluster().ensureAtLeastNumDataNodes(3);
1531+
1532+
allowToCompletePhase1Latch.countDown();
1533+
assertAcked(
1534+
client().admin()
1535+
.indices()
1536+
.prepareUpdateSettings(indexName)
1537+
.setSettings(
1538+
Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 2).putNull("index.routing.allocation.include._name")
1539+
)
1540+
);
1541+
assertFalse(client().admin().cluster().prepareHealth(indexName).setWaitForActiveShards(2).get().isTimedOut());
15441542
ensureGreen(indexName);
15451543
}
15461544

‎server/src/main/java/org/opensearch/cluster/routing/allocation/ExistingShardsAllocator.java‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -64,21 +64,19 @@ public interface ExistingShardsAllocator {
6464

6565
/**
6666
* Boolean setting to enable/disable batch allocation of unassigned shards already existing on disk.
67-
* This will allow sending all Unassigned Shards to the ExistingShard Allocator to make decision to allocate
67+
* This will allow sending all Unassigned Shards to the ExistingShard Allocator to make decision to allocate
6868
* in one or more go.
69-
*
70-
* Enable this setting if your ExistingShardAllocator is implementing the
69+
* <p>
70+
* This setting is enabled by default. In your ExistingShardAllocator implement the
7171
* {@link ExistingShardsAllocator#allocateAllUnassignedShards(RoutingAllocation, boolean)} method.
7272
* The default implementation of this method is not optimized and assigns shards one by one.
73-
*
74-
* If no plugin overrides {@link ExistingShardsAllocator} then default implementation will be use for it , i.e,
73+
* <p>
74+
* If no plugin overrides {@link ExistingShardsAllocator} then default implementation will be used for it , i.e,
7575
* {@link ShardsBatchGatewayAllocator}.
76-
*
77-
* This setting is experimental at this point.
7876
*/
7977
Setting<Boolean> EXISTING_SHARDS_ALLOCATOR_BATCH_MODE = Setting.boolSetting(
8078
"cluster.allocator.existing_shards_allocator.batch_enabled",
81-
false,
79+
true,
8280
Setting.Property.NodeScope
8381
);
8482

‎server/src/main/java/org/opensearch/gateway/ShardsBatchGatewayAllocator.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ public class ShardsBatchGatewayAllocator implements ExistingShardsAllocator {
8585
private TimeValue replicaShardsBatchGatewayAllocatorTimeout;
8686
private volatile Priority followUpRerouteTaskPriority;
8787
public static final TimeValue MIN_ALLOCATOR_TIMEOUT = TimeValue.timeValueSeconds(20);
88+
public static final TimeValue DEFAULT_ALLOCATOR_TIMEOUT = TimeValue.timeValueSeconds(20);
8889
private final ClusterManagerMetrics clusterManagerMetrics;
8990

9091
/**
@@ -105,7 +106,7 @@ public class ShardsBatchGatewayAllocator implements ExistingShardsAllocator {
105106
*/
106107
public static final Setting<TimeValue> PRIMARY_BATCH_ALLOCATOR_TIMEOUT_SETTING = Setting.timeSetting(
107108
PRIMARY_BATCH_ALLOCATOR_TIMEOUT_SETTING_KEY,
108-
TimeValue.MINUS_ONE,
109+
DEFAULT_ALLOCATOR_TIMEOUT,
109110
TimeValue.MINUS_ONE,
110111
new Setting.Validator<>() {
111112
@Override
@@ -129,7 +130,7 @@ public void validate(TimeValue timeValue) {
129130
*/
130131
public static final Setting<TimeValue> REPLICA_BATCH_ALLOCATOR_TIMEOUT_SETTING = Setting.timeSetting(
131132
REPLICA_BATCH_ALLOCATOR_TIMEOUT_SETTING_KEY,
132-
TimeValue.MINUS_ONE,
133+
DEFAULT_ALLOCATOR_TIMEOUT,
133134
TimeValue.MINUS_ONE,
134135
new Setting.Validator<>() {
135136
@Override

‎server/src/test/java/org/opensearch/cluster/routing/allocation/AllocationServiceTests.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,12 +59,14 @@
5959
import org.opensearch.common.settings.ClusterSettings;
6060
import org.opensearch.common.settings.Settings;
6161
import org.opensearch.gateway.GatewayAllocator;
62+
import org.opensearch.gateway.ShardsBatchGatewayAllocator;
6263
import org.opensearch.snapshots.EmptySnapshotsInfoService;
6364
import org.opensearch.telemetry.metrics.Histogram;
6465
import org.opensearch.telemetry.metrics.MetricsRegistry;
6566
import org.opensearch.telemetry.metrics.noop.NoopMetricsRegistry;
6667
import org.opensearch.test.OpenSearchTestCase;
6768
import org.opensearch.test.gateway.TestGatewayAllocator;
69+
import org.opensearch.test.gateway.TestShardBatchGatewayAllocator;
6870

6971
import java.util.Arrays;
7072
import java.util.Collections;
@@ -192,8 +194,10 @@ public ShardAllocationDecision decideShardAllocation(ShardRouting shard, Routing
192194
final String unrealisticAllocatorName = "unrealistic";
193195
final Map<String, ExistingShardsAllocator> allocatorMap = new HashMap<>();
194196
final TestGatewayAllocator testGatewayAllocator = new TestGatewayAllocator();
197+
final TestShardBatchGatewayAllocator testShardBatchGatewayAllocator = new TestShardBatchGatewayAllocator();
195198
allocatorMap.put(GatewayAllocator.ALLOCATOR_NAME, testGatewayAllocator);
196199
allocatorMap.put(unrealisticAllocatorName, new UnrealisticAllocator());
200+
allocatorMap.put(ShardsBatchGatewayAllocator.ALLOCATOR_NAME, testShardBatchGatewayAllocator);
197201
allocationService.setExistingShardsAllocators(allocatorMap);
198202

199203
final DiscoveryNodes.Builder nodesBuilder = DiscoveryNodes.builder();

‎server/src/test/java/org/opensearch/index/SearchSlowLogTests.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -211,15 +211,18 @@ public void testTwoLoggersDifferentLevel() {
211211
public void testMultipleSlowLoggersUseSingleLog4jLogger() {
212212
LoggerContext context = (LoggerContext) LogManager.getContext(false);
213213

214-
SearchContext ctx1 = searchContextWithSourceAndTask(createIndex("index-1"));
214+
IndexService index1 = createIndex("index-1");
215+
IndexService index2 = createIndex("index-2");
216+
217+
SearchContext ctx1 = searchContextWithSourceAndTask(index1);
215218
IndexSettings settings1 = new IndexSettings(
216219
createIndexMetadata(SlowLogLevel.WARN, "index-1", UUIDs.randomBase64UUID()),
217220
Settings.EMPTY
218221
);
219222
SearchSlowLog log1 = new SearchSlowLog(settings1);
220223
int numberOfLoggersBefore = context.getLoggers().size();
221224

222-
SearchContext ctx2 = searchContextWithSourceAndTask(createIndex("index-2"));
225+
SearchContext ctx2 = searchContextWithSourceAndTask(index2);
223226
IndexSettings settings2 = new IndexSettings(
224227
createIndexMetadata(SlowLogLevel.TRACE, "index-2", UUIDs.randomBase64UUID()),
225228
Settings.EMPTY

0 commit comments

Comments
 (0)