github-mirrors/logstash
1
0
Fork 0
mirror of https://github.com/elastic/logstash.git synced 2025-04-24 22:57:16 -04:00
Code Issues Projects Releases Packages Wiki Activity

release queue dir lock upon exceptions while opening queue

Browse source
This commit is contained in:
Colin Surprenant 2020-06-16 16:53:49 -04:00
parent 8eddbd608b
commit c2e4e4f185
2 changed files with 161 additions and 123 deletions
Download patch file Download diff file Expand all files Collapse all files

45
logstash-core/src/main/java/org/logstash/ackedqueue/Queue.java
View file

@ -158,14 +158,34 @@ public final class Queue implements Closeable {
* @throws IOException if an IO error occurs
*/
public void open() throws IOException {
final int headPageNum;
if (!this.closed.get()) { throw new IOException("queue already opened"); }
lock.lock();
try {
try {
// verify exclusive access to the dirPath
this.dirLock = FileLockFactory.obtainLock(this.dirPath, LOCK_NAME);
} catch (LockException e) {
throw new LockException("The queue failed to obtain exclusive access, cause: " + e.getMessage());
}
try {
openPages();
this.closed.set(false);
} catch (IOException e) {
// upon any exception while opening the queue and after dirlock has been obtained
// we need to make sure to release the dirlock. Calling the close method on a partially
// open queue has no effect because the closed flag is still true.
releaseLockAndSwallow();
throw(e);
}
} finally {
lock.unlock();
}
}
private void openPages() throws IOException {
final int headPageNum;
// Upgrade to serialization format V2
QueueUpgrade.upgradeQueueDirectoryToV2(dirPath);
@ -229,7 +249,7 @@ public final class Queue implements Closeable {
PageIO pageIO = new MmapPageIOV2(headCheckpoint.getPageNum(), this.pageCapacity, this.dirPath);
pageIO.recover(); // optimistically recovers the head page data file and set minSeqNum and elementCount to the actual read/recovered data
pqSizeBytes += (long)pageIO.getHead();
pqSizeBytes += (long) pageIO.getHead();
ensureDiskAvailable(this.maxBytes, pqSizeBytes);
if (pageIO.getMinSeqNum() != headCheckpoint.getMinSeqNum() || pageIO.getElementCount() != headCheckpoint.getElementCount()) {
@ -277,14 +297,8 @@ public final class Queue implements Closeable {
}
// TODO: here do directory traversal and cleanup lingering pages? could be a background operations to not delay queue start?
}
this.closed.set(false);
} catch (LockException e) {
throw new LockException("The queue failed to obtain exclusive access, cause: " + e.getMessage());
} finally {
lock.unlock();
}
}
/**
* delete files for the given page
@ -713,15 +727,18 @@ public final class Queue implements Closeable {
notFull.signalAll();
} finally {
releaseLockAndSwallow();
lock.unlock();
}
}
}
private void releaseLockAndSwallow() {
try {
FileLockFactory.releaseLock(this.dirLock);
} catch (IOException e) {
// log error and ignore
logger.error("Queue close releaseLock failed, error={}", e.getMessage());
} finally {
lock.unlock();
}
}
}
}

21
logstash-core/src/test/java/org/logstash/ackedqueue/QueueTest.java
View file

@ -38,6 +38,8 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import org.hamcrest.CoreMatchers;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
@ -1043,4 +1045,23 @@ public class QueueTest {
queue.open();
}
}
@Test
public void lockIsReleasedUponOpenException() throws Exception {
Settings settings = SettingsImpl.builder(TestSettings.persistedQueueSettings(100, dataPath))
.queueMaxBytes(Long.MAX_VALUE)
.build();
try {
Queue queue = new Queue(settings);
queue.open();
fail("expected queue.open() to throws when not enough disk free");
} catch (IOException e) {
assertThat(e.getMessage(), CoreMatchers.containsString("Unable to allocate"));
}
// at this point the Queue lock should be released and Queue.open should not throw a LockException
try (Queue queue = new Queue(TestSettings.persistedQueueSettings(10, dataPath))) {
queue.open();
}
}
}