honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-02-17 10:25:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

[ML] Fix thread leak when waiting for job flush (#32196) (#32541)

Browse Source
This commit is contained in:
Benjamin Trent 2018-08-01 10:38:04 -05:00 committed by GitHub
parent 7a3eb4b3e8
commit 9fb790dcc3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 21 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectCommunicator.java
View File

@ -270,7 +270,7 @@ public class AutodetectCommunicator implements Closeable {
}
@Nullable
FlushAcknowledgement waitFlushToCompletion(String flushId) {
FlushAcknowledgement waitFlushToCompletion(String flushId) throws InterruptedException {
LOGGER.debug("[{}] waiting for flush", job.getId());
FlushAcknowledgement flushAcknowledgement;

2
x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/output/AutoDetectResultProcessor.java
View File

@ -485,7 +485,7 @@ public class AutoDetectResultProcessor {
* @return The {@link FlushAcknowledgement} if the flush has completed or the parsing finished; {@code null} if the timeout expired
*/
@Nullable
public FlushAcknowledgement waitForFlushAcknowledgement(String flushId, Duration timeout) {
public FlushAcknowledgement waitForFlushAcknowledgement(String flushId, Duration timeout) throws InterruptedException {
return failed ? null : flushListener.waitForFlush(flushId, timeout);
}

10
x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/output/FlushListener.java
View File

@ -22,18 +22,14 @@ class FlushListener {
final AtomicBoolean cleared = new AtomicBoolean(false);
@Nullable
FlushAcknowledgement waitForFlush(String flushId, Duration timeout) {
FlushAcknowledgement waitForFlush(String flushId, Duration timeout) throws InterruptedException {
if (cleared.get()) {
return null;
}
FlushAcknowledgementHolder holder = awaitingFlushed.computeIfAbsent(flushId, (key) -> new FlushAcknowledgementHolder(flushId));
try {
if (holder.latch.await(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
return holder.flushAcknowledgement;
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
if (holder.latch.await(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
return holder.flushAcknowledgement;
}
return null;
}

6
x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/AutodetectCommunicatorTests.java
View File

@ -108,7 +108,7 @@ public class AutodetectCommunicatorTests extends ESTestCase {
verifyNoMoreInteractions(process);
}
public void testFlushJob() throws IOException {
public void testFlushJob() throws IOException, InterruptedException {
AutodetectProcess process = mockAutodetectProcessWithOutputStream();
when(process.isProcessAlive()).thenReturn(true);
AutoDetectResultProcessor processor = mock(AutoDetectResultProcessor.class);
@ -123,7 +123,7 @@ public class AutodetectCommunicatorTests extends ESTestCase {
}
}
public void testWaitForFlushReturnsIfParserFails() throws IOException {
public void testWaitForFlushReturnsIfParserFails() throws IOException, InterruptedException {
AutodetectProcess process = mockAutodetectProcessWithOutputStream();
when(process.isProcessAlive()).thenReturn(true);
AutoDetectResultProcessor processor = mock(AutoDetectResultProcessor.class);
@ -144,7 +144,7 @@ public class AutodetectCommunicatorTests extends ESTestCase {
assertEquals("[foo] Unexpected death of autodetect: Mock process is dead", holder[0].getMessage());
}
public void testFlushJob_givenFlushWaitReturnsTrueOnSecondCall() throws IOException {
public void testFlushJob_givenFlushWaitReturnsTrueOnSecondCall() throws IOException, InterruptedException {
AutodetectProcess process = mockAutodetectProcessWithOutputStream();
when(process.isProcessAlive()).thenReturn(true);
AutoDetectResultProcessor autoDetectResultProcessor = Mockito.mock(AutoDetectResultProcessor.class);

2
x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/output/AutoDetectResultProcessorTests.java
View File

@ -514,7 +514,7 @@ public class AutoDetectResultProcessorTests extends ESTestCase {
verify(persister, times(2)).persistModelSnapshot(any(), eq(WriteRequest.RefreshPolicy.IMMEDIATE));
}
public void testParsingErrorSetsFailed() {
public void testParsingErrorSetsFailed() throws InterruptedException {
@SuppressWarnings("unchecked")
Iterator<AutodetectResult> iterator = mock(Iterator.class);
when(iterator.hasNext()).thenThrow(new ElasticsearchParseException("this test throws"));

16
x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/output/FlushListenerTests.java
View File

@ -22,8 +22,12 @@ public class FlushListenerTests extends ESTestCase {
FlushListener listener = new FlushListener();
AtomicReference<FlushAcknowledgement> flushAcknowledgementHolder = new AtomicReference<>();
new Thread(() -> {
FlushAcknowledgement flushAcknowledgement = listener.waitForFlush("_id", Duration.ofMillis(10000));
flushAcknowledgementHolder.set(flushAcknowledgement);
try {
FlushAcknowledgement flushAcknowledgement = listener.waitForFlush("_id", Duration.ofMillis(10000));
flushAcknowledgementHolder.set(flushAcknowledgement);
} catch (InterruptedException _ex) {
Thread.currentThread().interrupt();
}
}).start();
assertBusy(() -> assertTrue(listener.awaitingFlushed.containsKey("_id")));
assertNull(flushAcknowledgementHolder.get());
@ -46,8 +50,12 @@ public class FlushListenerTests extends ESTestCase {
AtomicReference<FlushAcknowledgement> flushAcknowledgementHolder = new AtomicReference<>();
flushAcknowledgementHolders.add(flushAcknowledgementHolder);
new Thread(() -> {
FlushAcknowledgement flushAcknowledgement = listener.waitForFlush(String.valueOf(id), Duration.ofMillis(10000));
flushAcknowledgementHolder.set(flushAcknowledgement);
try {
FlushAcknowledgement flushAcknowledgement = listener.waitForFlush(String.valueOf(id), Duration.ofMillis(10000));
flushAcknowledgementHolder.set(flushAcknowledgement);
} catch (InterruptedException _ex) {
Thread.currentThread().interrupt();
}
}).start();
}
assertBusy(() -> assertEquals(numWaits, listener.awaitingFlushed.size()));