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

#7399 stress test full queue situation

Browse source 
Fixes #7405
This commit is contained in:
Armin 2017-06-09 20:00:44 +02:00 • committed by Armin Braun
parent f5c6c5a4b1
commit 2ab35b61ea
1 changed files with 54 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -563,51 +563,13 @@ public class QueueTest {
}
@Test
public void queueStableUnderStress() throws Exception {
Settings settings = TestSettings.persistedQueueSettings(1000000, dataPath);
final ExecutorService exec = Executors.newScheduledThreadPool(2);
try (Queue queue = new Queue(settings)) {
final int count = 20_000;
final int concurrent = 2;
queue.open();
final Future<Integer>[] futures = new Future[concurrent];
for (int c = 0; c < concurrent; ++c) {
futures[c] = exec.submit(() -> {
int i = 0;
try {
while (i < count / concurrent) {
final Batch batch = queue.readBatch(1);
for (final Queueable elem : batch.getElements()) {
if (elem != null) {
++i;
}
}
}
return i;
} catch (final IOException ex) {
throw new IllegalStateException(ex);
}
});
}
for (int i = 0; i < count; ++i) {
try {
final Queueable evnt = new StringElement("foo");
queue.write(evnt);
} catch (final IOException ex) {
throw new IllegalStateException(ex);
}
}
assertThat(
Arrays.stream(futures).map(i -> {
try {
return i.get(10L, TimeUnit.SECONDS);
} catch (final InterruptedException | ExecutionException | TimeoutException ex) {
throw new IllegalStateException(ex);
}
}).reduce((x, y) -> x + y).orElse(0),
is(20_000)
);
}
public void queueStableUnderStressHugeCapacity() throws Exception {
stableUnderStress(100_000);
}
@Test
public void queueStableUnderStressLowCapacity() throws Exception {
stableUnderStress(50);
}
@Test
@ -755,4 +717,51 @@ public class QueueTest {
assertThat(q.getPersistedByteSize(), is(0L));
}
}
private void stableUnderStress(final int capacity) throws IOException {
Settings settings = TestSettings.persistedQueueSettings(capacity, dataPath);
final ExecutorService exec = Executors.newScheduledThreadPool(2);
try (Queue queue = new Queue(settings)) {
final int count = 20_000;
final int concurrent = 2;
queue.open();
final Future<Integer>[] futures = new Future[concurrent];
for (int c = 0; c < concurrent; ++c) {
futures[c] = exec.submit(() -> {
int i = 0;
try {
while (i < count / concurrent) {
final Batch batch = queue.readBatch(1);
for (final Queueable elem : batch.getElements()) {
if (elem != null) {
++i;
}
}
}
return i;
} catch (final IOException ex) {
throw new IllegalStateException(ex);
}
});
}
for (int i = 0; i < count; ++i) {
try {
final Queueable evnt = new StringElement("foo");
queue.write(evnt);
} catch (final IOException ex) {
throw new IllegalStateException(ex);
}
}
assertThat(
Arrays.stream(futures).map(i -> {
try {
return i.get(2L, TimeUnit.MINUTES);
} catch (final InterruptedException | ExecutionException | TimeoutException ex) {
throw new IllegalStateException(ex);
}
}).reduce((x, y) -> x + y).orElse(0),
is(20_000)
);
}
}
}