Skip to content

Commit e6d7ca6

Browse files
LantaoJinLantaoJin
committed
Trigger GC when catch MemoryUsageExceedException on the first retry
Signed-off-by: Lantao Jin <ltjin@amazon.com>
1 parent 0f197e5 commit e6d7ca6

4 files changed

Lines changed: 69 additions & 44 deletions

File tree

‎opensearch/src/main/java/org/opensearch/sql/opensearch/monitor/OpenSearchMemoryHealthy.java‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -6,24 +6,23 @@
66
package org.opensearch.sql.opensearch.monitor;
77

88
import com.google.common.annotations.VisibleForTesting;
9-
import java.util.concurrent.ThreadLocalRandom;
109
import lombok.NoArgsConstructor;
1110
import lombok.extern.log4j.Log4j2;
1211

1312
/** OpenSearch Memory Monitor. */
1413
@Log4j2
1514
public class OpenSearchMemoryHealthy {
16-
private final RandomFail randomFail;
15+
private final FastFail fastFail;
1716
private final MemoryUsage memoryUsage;
1817

1918
public OpenSearchMemoryHealthy() {
20-
randomFail = new RandomFail();
19+
fastFail = new FastFail();
2120
memoryUsage = new MemoryUsage();
2221
}
2322

2423
@VisibleForTesting
25-
public OpenSearchMemoryHealthy(RandomFail randomFail, MemoryUsage memoryUsage) {
26-
this.randomFail = randomFail;
24+
public OpenSearchMemoryHealthy(FastFail fastFail, MemoryUsage memoryUsage) {
25+
this.fastFail = fastFail;
2726
this.memoryUsage = memoryUsage;
2827
}
2928

@@ -35,7 +34,8 @@ public boolean isMemoryHealthy(long limitBytes) {
3534
return true;
3635
} else {
3736
log.warn("Memory usage:{} exceed limit:{}", memoryUsage, limitBytes);
38-
if (randomFail.shouldFail()) {
37+
// always return false in prod, only can be true in testing
38+
if (fastFail.shouldFail()) {
3939
log.warn("Fast failing the current request");
4040
throw new MemoryUsageExceedFastFailureException();
4141
} else {
@@ -44,9 +44,9 @@ public boolean isMemoryHealthy(long limitBytes) {
4444
}
4545
}
4646

47-
static class RandomFail {
47+
static class FastFail {
4848
public boolean shouldFail() {
49-
return ThreadLocalRandom.current().nextBoolean();
49+
return false;
5050
}
5151
}
5252

‎opensearch/src/main/java/org/opensearch/sql/opensearch/monitor/OpenSearchResourceMonitor.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,15 @@ public OpenSearchResourceMonitor(Settings settings, OpenSearchMemoryHealthy memo
3636
.ignoreExceptions(OpenSearchMemoryHealthy.MemoryUsageExceedFastFailureException.class)
3737
.build();
3838
retry = Retry.of("mem", config);
39+
retry
40+
.getEventPublisher()
41+
.onRetry(
42+
event -> {
43+
if (event.getNumberOfRetryAttempts() == 1) {
44+
System.gc();
45+
log.warn("isMemoryHealthy() failed in first retry, triggered System.gc()");
46+
}
47+
});
3948
this.memoryMonitor = memoryMonitor;
4049
}
4150

‎opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java‎

Lines changed: 46 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
package org.opensearch.sql.opensearch.storage;
77

88
import com.google.common.annotations.VisibleForTesting;
9+
import java.lang.ref.SoftReference;
910
import java.util.HashMap;
1011
import java.util.LinkedHashMap;
1112
import java.util.Map;
@@ -48,7 +49,6 @@
4849

4950
/** OpenSearch table (index) implementation. */
5051
public class OpenSearchIndex extends AbstractOpenSearchTable {
51-
5252
public static final String METADATA_FIELD_ID = "_id";
5353
public static final String METADATA_FIELD_INDEX = "_index";
5454
public static final String METADATA_FIELD_SCORE = "_score";
@@ -77,14 +77,23 @@ public class OpenSearchIndex extends AbstractOpenSearchTable {
7777
/** {@link OpenSearchRequest.IndexName}. */
7878
private final OpenSearchRequest.IndexName indexName;
7979

80-
/** The cached mapping of field and type in index. */
81-
private Map<String, OpenSearchDataType> cachedFieldOpenSearchTypes = null;
80+
/**
81+
* The cached mapping of field and type in index. Do not call cachedFieldOpenSearchTypes.get()
82+
* directly, use {@link #getFieldOpenSearchTypes() instead }
83+
*/
84+
private SoftReference<Map<String, OpenSearchDataType>> cachedFieldOpenSearchTypes = null;
8285

83-
/** The cached ExprType of fields. */
84-
private Map<String, ExprType> cachedFieldTypes = null;
86+
/**
87+
* The cached ExprType of fields. Do not call cachedFieldTypes.get() directly, use {@link
88+
* #getFieldTypes() instead }
89+
*/
90+
private SoftReference<Map<String, ExprType>> cachedFieldTypes = null;
8591

86-
/** The cached mapping of alias type field to its original path. */
87-
private Map<String, String> aliasMapping = null;
92+
/**
93+
* The cached mapping of alias type field to its original path. Do not call aliasMapping.get()
94+
* directly, use {@link #getAliasMapping() instead }
95+
*/
96+
private SoftReference<Map<String, String>> aliasMapping = null;
8897

8998
/** The cached max result window setting of index. */
9099
private Integer cachedMaxResultWindow = null;
@@ -132,19 +141,23 @@ public void create(Map<String, ExprType> schema) {
132141
*/
133142
@Override
134143
public Map<String, ExprType> getFieldTypes() {
135-
if (cachedFieldOpenSearchTypes == null) {
144+
if (cachedFieldOpenSearchTypes == null || cachedFieldOpenSearchTypes.get() == null) {
136145
cachedFieldOpenSearchTypes =
137-
new OpenSearchDescribeIndexRequest(client, indexName).getFieldTypes();
146+
new SoftReference<>(
147+
new OpenSearchDescribeIndexRequest(client, indexName).getFieldTypes());
138148
}
139-
if (cachedFieldTypes == null) {
149+
if (cachedFieldTypes == null || cachedFieldTypes.get() == null) {
140150
cachedFieldTypes =
141-
OpenSearchDataType.traverseAndFlatten(cachedFieldOpenSearchTypes).entrySet().stream()
142-
.collect(
143-
LinkedHashMap::new,
144-
(map, item) -> map.put(item.getKey(), item.getValue().getExprType()),
145-
Map::putAll);
151+
new SoftReference<>(
152+
OpenSearchDataType.traverseAndFlatten(cachedFieldOpenSearchTypes.get())
153+
.entrySet()
154+
.stream()
155+
.collect(
156+
LinkedHashMap::new,
157+
(map, item) -> map.put(item.getKey(), item.getValue().getExprType()),
158+
Map::putAll));
146159
}
147-
return cachedFieldTypes;
160+
return cachedFieldTypes.get();
148161
}
149162

150163
@Override
@@ -153,19 +166,21 @@ public Map<String, ExprType> getReservedFieldTypes() {
153166
}
154167

155168
public Map<String, String> getAliasMapping() {
156-
if (cachedFieldOpenSearchTypes == null) {
169+
if (cachedFieldOpenSearchTypes == null || cachedFieldOpenSearchTypes.get() == null) {
157170
cachedFieldOpenSearchTypes =
158-
new OpenSearchDescribeIndexRequest(client, indexName).getFieldTypes();
171+
new SoftReference<>(
172+
new OpenSearchDescribeIndexRequest(client, indexName).getFieldTypes());
159173
}
160-
if (aliasMapping == null) {
174+
if (aliasMapping == null || aliasMapping.get() == null) {
161175
aliasMapping =
162-
cachedFieldOpenSearchTypes.entrySet().stream()
163-
.filter(entry -> entry.getValue().getOriginalPath().isPresent())
164-
.collect(
165-
Collectors.toUnmodifiableMap(
166-
Entry::getKey, entry -> entry.getValue().getOriginalPath().get()));
176+
new SoftReference<>(
177+
cachedFieldOpenSearchTypes.get().entrySet().stream()
178+
.filter(entry -> entry.getValue().getOriginalPath().isPresent())
179+
.collect(
180+
Collectors.toUnmodifiableMap(
181+
Entry::getKey, entry -> entry.getValue().getOriginalPath().get())));
167182
}
168-
return aliasMapping;
183+
return aliasMapping.get();
169184
}
170185

171186
/**
@@ -174,11 +189,12 @@ public Map<String, String> getAliasMapping() {
174189
* @return A complete map between field names and their types.
175190
*/
176191
public Map<String, OpenSearchDataType> getFieldOpenSearchTypes() {
177-
if (cachedFieldOpenSearchTypes == null) {
192+
if (cachedFieldOpenSearchTypes == null || cachedFieldOpenSearchTypes.get() == null) {
178193
cachedFieldOpenSearchTypes =
179-
new OpenSearchDescribeIndexRequest(client, indexName).getFieldTypes();
194+
new SoftReference<>(
195+
new OpenSearchDescribeIndexRequest(client, indexName).getFieldTypes());
180196
}
181-
return cachedFieldOpenSearchTypes;
197+
return cachedFieldOpenSearchTypes.get();
182198
}
183199

184200
/** Get the max result window setting of the table. */
@@ -207,7 +223,7 @@ public TableScanBuilder createScanBuilder() {
207223
client,
208224
requestBuilder.getMaxResponseSize(),
209225
requestBuilder.build(
210-
indexName, cursorKeepAlive, client, cachedFieldOpenSearchTypes.isEmpty()));
226+
indexName, cursorKeepAlive, client, getFieldOpenSearchTypes().isEmpty()));
211227
return new OpenSearchIndexScanBuilder(builder, createScanOperator);
212228
}
213229

@@ -266,6 +282,6 @@ public OpenSearchResourceMonitor createOpenSearchResourceMonitor() {
266282
public OpenSearchRequest buildRequest(OpenSearchRequestBuilder requestBuilder) {
267283
final TimeValue cursorKeepAlive = settings.getSettingValue(Settings.Key.SQL_CURSOR_KEEP_ALIVE);
268284
return requestBuilder.build(
269-
indexName, cursorKeepAlive, client, cachedFieldOpenSearchTypes.isEmpty());
285+
indexName, cursorKeepAlive, client, getFieldOpenSearchTypes().isEmpty());
270286
}
271287
}

‎opensearch/src/test/java/org/opensearch/sql/opensearch/monitor/OpenSearchMemoryHealthyTest.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -19,15 +19,15 @@
1919
@ExtendWith(MockitoExtension.class)
2020
class OpenSearchMemoryHealthyTest {
2121

22-
@Mock private OpenSearchMemoryHealthy.RandomFail randomFail;
22+
@Mock private OpenSearchMemoryHealthy.FastFail fastFail;
2323

2424
@Mock private OpenSearchMemoryHealthy.MemoryUsage memoryUsage;
2525

2626
private OpenSearchMemoryHealthy monitor;
2727

2828
@BeforeEach
2929
public void setup() {
30-
monitor = new OpenSearchMemoryHealthy(randomFail, memoryUsage);
30+
monitor = new OpenSearchMemoryHealthy(fastFail, memoryUsage);
3131
}
3232

3333
@Test
@@ -40,7 +40,7 @@ void isMemoryHealthy() {
4040
@Test
4141
void memoryUsageExceedLimitFastFailure() {
4242
when(memoryUsage.usage()).thenReturn(10L);
43-
when(randomFail.shouldFail()).thenReturn(true);
43+
when(fastFail.shouldFail()).thenReturn(true);
4444

4545
assertThrows(
4646
OpenSearchMemoryHealthy.MemoryUsageExceedFastFailureException.class,
@@ -50,7 +50,7 @@ void memoryUsageExceedLimitFastFailure() {
5050
@Test
5151
void memoryUsageExceedLimitWithoutFastFailure() {
5252
when(memoryUsage.usage()).thenReturn(10L);
53-
when(randomFail.shouldFail()).thenReturn(false);
53+
when(fastFail.shouldFail()).thenReturn(false);
5454

5555
assertThrows(
5656
OpenSearchMemoryHealthy.MemoryUsageExceedException.class,
@@ -65,8 +65,8 @@ void constructOpenSearchMemoryMonitorWithoutArguments() {
6565

6666
@Test
6767
void randomFail() {
68-
OpenSearchMemoryHealthy.RandomFail randomFail = new OpenSearchMemoryHealthy.RandomFail();
69-
assertNotNull(randomFail.shouldFail());
68+
OpenSearchMemoryHealthy.FastFail fastFail = new OpenSearchMemoryHealthy.FastFail();
69+
assertNotNull(fastFail.shouldFail());
7070
}
7171

7272
@Test

0 commit comments

Comments
 (0)