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

Change Remote state read thread pool to Fixed type #16850

Merged
merged 2 commits into from
Dec 18, 2024
Merged

Change Remote state read thread pool to Fixed type #16850

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1c55c53
Change Remote state read thread pool to Fixed type
soosinha Dec 16, 2024
ab77523
Add more coverage
soosinha Dec 17, 2024
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
9 changes: 8 additions & 1 deletion .../java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -430,6 +430,13 @@

if (remoteClusterStateService != null && termVersionResponse.isStatePresentInRemote()) {
try {
logger.info(
() -> new ParameterizedMessage(
"Term version checker downloading full cluster state for term {}, version {}",
termVersion.getTerm(),
termVersion.getVersion()
)
);
ClusterStateTermVersion clusterStateTermVersion = termVersionResponse.getClusterStateTermVersion();
Optional<ClusterMetadataManifest> clusterMetadataManifest = remoteClusterStateService
.getClusterMetadataManifestByTermVersion(
Expand All @@ -454,7 +461,7 @@
return clusterStateFromRemote;
}
} catch (Exception e) {
logger.trace("Error while fetching from remote cluster state", e);
logger.error("Error while fetching from remote cluster state", e);

Check warning on line 464 in server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeAction.java

View check run for this annotation

Codecov / codecov/patch

server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeAction.java#L464 

Added line #L464 was not covered by tests
}
}
return null;
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -258,7 +258,7 @@ PublishWithJoinResponse handleIncomingRemotePublishRequest(RemotePublishRequest
}

if (applyFullState == true) {
logger.debug(
logger.info(
() -> new ParameterizedMessage(
"Downloading full cluster state for term {}, version {}, stateUUID {}",
manifest.getClusterTerm(),
Expand Down
14 changes: 14 additions & 0 deletions server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1473,8 +1473,22 @@
try {
ClusterState stateFromCache = remoteClusterStateCache.getState(clusterName, manifest);
if (stateFromCache != null) {
logger.trace(
() -> new ParameterizedMessage(

Check warning on line 1477 in server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java

View check run for this annotation

Codecov / codecov/patch

server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java#L1476-L1477 

Added lines #L1476 - L1477 were not covered by tests
"Found cluster state in cache for term {} and version {}",
manifest.getClusterTerm(),
manifest.getStateVersion()

Check warning on line 1480 in server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java

View check run for this annotation

Codecov / codecov/patch

server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java#L1479-L1480 

Added lines #L1479 - L1480 were not covered by tests
)
);
return stateFromCache;
}
logger.info(
() -> new ParameterizedMessage(
"Cluster state not found in cache for term {} and version {}",
manifest.getClusterTerm(),
manifest.getStateVersion()
)
);

final ClusterState clusterState;
final long startTimeNanos = relativeTimeNanosSupplier.getAsLong();
Expand Down
4 changes: 2 additions & 2 deletions server/src/main/java/org/opensearch/threadpool/ThreadPool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,7 +198,7 @@ public static ThreadPoolType fromType(String type) {
map.put(Names.REMOTE_PURGE, ThreadPoolType.SCALING);
map.put(Names.REMOTE_REFRESH_RETRY, ThreadPoolType.SCALING);
map.put(Names.REMOTE_RECOVERY, ThreadPoolType.SCALING);
map.put(Names.REMOTE_STATE_READ, ThreadPoolType.SCALING);
map.put(Names.REMOTE_STATE_READ, ThreadPoolType.FIXED);
map.put(Names.INDEX_SEARCHER, ThreadPoolType.RESIZABLE);
map.put(Names.REMOTE_STATE_CHECKSUM, ThreadPoolType.FIXED);
THREAD_POOL_TYPES = Collections.unmodifiableMap(map);
Expand Down Expand Up @@ -306,7 +306,7 @@ public ThreadPool(
);
builders.put(
Names.REMOTE_STATE_READ,
new ScalingExecutorBuilder(Names.REMOTE_STATE_READ, 1, boundedBy(4 * allocatedProcessors, 4, 32), TimeValue.timeValueMinutes(5))
new FixedExecutorBuilder(settings, Names.REMOTE_STATE_READ, boundedBy(4 * allocatedProcessors, 4, 32), 120000)
soosinha marked this conversation as resolved.
Show resolved Hide resolved
);
builders.put(
Names.INDEX_SEARCHER,
Expand Down
1 change: 0 additions & 1 deletion server/src/test/java/org/opensearch/threadpool/ScalingThreadPoolTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,7 +156,6 @@ private int expectedSize(final String threadPoolName, final int numberOfProcesso
sizes.put(ThreadPool.Names.REMOTE_PURGE, ThreadPool::halfAllocatedProcessors);
sizes.put(ThreadPool.Names.REMOTE_REFRESH_RETRY, ThreadPool::halfAllocatedProcessors);
sizes.put(ThreadPool.Names.REMOTE_RECOVERY, ThreadPool::twiceAllocatedProcessors);
sizes.put(ThreadPool.Names.REMOTE_STATE_READ, n -> ThreadPool.boundedBy(4 * n, 4, 32));
return sizes.get(threadPoolName).apply(numberOfProcessors);
}

Expand Down
Loading