Skip to content

Commit 98106dd

Browse files
adinauerclaude
adinauer
and
claude
authored
fix(core): Prevent lost wakeups in batch processors (#5756)
* fix(core): Prevent lost wakeups in batch processors Coordinate log and metric flush scheduling through an atomic state transition. This ensures events added while a flush completes always have a pending task instead of relying on the running Future's completion state. Fixes #5739 Co-Authored-By: Claude <noreply@anthropic.com> * changelog --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent 20be537 commit 98106dd

5 files changed

Lines changed: 92 additions & 68 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030

3131
### Fixes
3232

33+
- Prevent logs and metrics from remaining queued after a flush scheduling race ([#5756](https://github.com/getsentry/sentry-java/pull/5756))
3334
- Fix main thread identification for tombstone (native crash) events ([#5742](https://github.com/getsentry/sentry-java/pull/5742))
3435

3536
### Dependencies

‎sentry/src/main/java/io/sentry/logger/LoggerBatchProcessor.java‎

Lines changed: 22 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -4,23 +4,21 @@
44
import io.sentry.DataCategory;
55
import io.sentry.ISentryClient;
66
import io.sentry.ISentryExecutorService;
7-
import io.sentry.ISentryLifecycleToken;
87
import io.sentry.SentryExecutorService;
98
import io.sentry.SentryLevel;
109
import io.sentry.SentryLogEvent;
1110
import io.sentry.SentryLogEvents;
1211
import io.sentry.SentryOptions;
1312
import io.sentry.clientreport.DiscardReason;
1413
import io.sentry.transport.ReusableCountLatch;
15-
import io.sentry.util.AutoClosableReentrantLock;
1614
import io.sentry.util.JsonSerializationUtils;
1715
import java.util.ArrayList;
1816
import java.util.List;
1917
import java.util.Queue;
2018
import java.util.concurrent.ConcurrentLinkedQueue;
21-
import java.util.concurrent.Future;
2219
import java.util.concurrent.RejectedExecutionException;
2320
import java.util.concurrent.TimeUnit;
21+
import java.util.concurrent.atomic.AtomicBoolean;
2422
import org.jetbrains.annotations.ApiStatus;
2523
import org.jetbrains.annotations.NotNull;
2624
import org.jetbrains.annotations.Nullable;
@@ -37,9 +35,7 @@ public class LoggerBatchProcessor implements ILoggerBatchProcessor {
3735
private final @NotNull ISentryClient client;
3836
private final @NotNull Queue<SentryLogEvent> queue;
3937
private final @NotNull ISentryExecutorService executorService;
40-
private volatile @Nullable Future<?> scheduledFlush;
41-
private final @NotNull AutoClosableReentrantLock scheduleLock = new AutoClosableReentrantLock();
42-
private volatile boolean hasScheduled = false;
38+
private final @NotNull AtomicBoolean hasScheduled = new AtomicBoolean(false);
4339
private volatile boolean isShuttingDown = false;
4440

4541
private final @NotNull ReusableCountLatch pendingCount = new ReusableCountLatch();
@@ -79,15 +75,15 @@ public void add(final @NotNull SentryLogEvent logEvent) {
7975
}
8076
pendingCount.increment();
8177
queue.offer(logEvent);
82-
maybeSchedule(false, false);
78+
maybeSchedule(false);
8379
}
8480

8581
@SuppressWarnings("FutureReturnValueIgnored")
8682
@Override
8783
public void close(final boolean isRestarting) {
8884
isShuttingDown = true;
8985
if (isRestarting) {
90-
maybeSchedule(true, true);
86+
maybeSchedule(true);
9187
executorService.submit(() -> executorService.close(options.getShutdownTimeoutMillis()));
9288
} else {
9389
executorService.close(options.getShutdownTimeoutMillis());
@@ -97,33 +93,28 @@ public void close(final boolean isRestarting) {
9793
}
9894
}
9995

100-
private void maybeSchedule(boolean forceSchedule, boolean immediately) {
101-
if (hasScheduled && !forceSchedule) {
96+
@SuppressWarnings("FutureReturnValueIgnored")
97+
private void maybeSchedule(boolean immediately) {
98+
if (immediately) {
99+
// any already scheduled task may be far in the future, we want to schedule something that
100+
// runs right away
101+
hasScheduled.set(true);
102+
} else if (!hasScheduled.compareAndSet(false, true)) {
103+
// was already true, no need to schedule again
102104
return;
103105
}
104-
try (final @NotNull ISentryLifecycleToken ignored = scheduleLock.acquire()) {
105-
final @Nullable Future<?> latestScheduledFlush = scheduledFlush;
106-
if (forceSchedule
107-
|| latestScheduledFlush == null
108-
|| latestScheduledFlush.isDone()
109-
|| latestScheduledFlush.isCancelled()) {
110-
hasScheduled = true;
111-
final int flushAfterMs = immediately ? 0 : FLUSH_AFTER_MS;
112-
try {
113-
scheduledFlush = executorService.schedule(new BatchRunnable(), flushAfterMs);
114-
} catch (RejectedExecutionException e) {
115-
hasScheduled = false;
116-
options
117-
.getLogger()
118-
.log(SentryLevel.WARNING, "Logs batch processor flush task rejected", e);
119-
}
120-
}
106+
final int flushAfterMs = immediately ? 0 : FLUSH_AFTER_MS;
107+
try {
108+
executorService.schedule(new BatchRunnable(), flushAfterMs);
109+
} catch (RejectedExecutionException e) {
110+
hasScheduled.set(false);
111+
options.getLogger().log(SentryLevel.WARNING, "Logs batch processor flush task rejected", e);
121112
}
122113
}
123114

124115
@Override
125116
public void flush(long timeoutMillis) {
126-
maybeSchedule(true, true);
117+
maybeSchedule(true);
127118
try {
128119
pendingCount.waitTillZero(timeoutMillis, TimeUnit.MILLISECONDS);
129120
} catch (InterruptedException e) {
@@ -134,12 +125,9 @@ public void flush(long timeoutMillis) {
134125

135126
private void flush() {
136127
flushInternal();
137-
try (final @NotNull ISentryLifecycleToken ignored = scheduleLock.acquire()) {
138-
if (!queue.isEmpty()) {
139-
maybeSchedule(true, false);
140-
} else {
141-
hasScheduled = false;
142-
}
128+
hasScheduled.set(false);
129+
if (!queue.isEmpty()) {
130+
maybeSchedule(false);
143131
}
144132
}
145133

‎sentry/src/main/java/io/sentry/metrics/MetricsBatchProcessor.java‎

Lines changed: 24 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -4,23 +4,21 @@
44
import io.sentry.DataCategory;
55
import io.sentry.ISentryClient;
66
import io.sentry.ISentryExecutorService;
7-
import io.sentry.ISentryLifecycleToken;
87
import io.sentry.SentryExecutorService;
98
import io.sentry.SentryLevel;
109
import io.sentry.SentryMetricsEvent;
1110
import io.sentry.SentryMetricsEvents;
1211
import io.sentry.SentryOptions;
1312
import io.sentry.clientreport.DiscardReason;
1413
import io.sentry.transport.ReusableCountLatch;
15-
import io.sentry.util.AutoClosableReentrantLock;
1614
import io.sentry.util.JsonSerializationUtils;
1715
import java.util.ArrayList;
1816
import java.util.List;
1917
import java.util.Queue;
2018
import java.util.concurrent.ConcurrentLinkedQueue;
21-
import java.util.concurrent.Future;
2219
import java.util.concurrent.RejectedExecutionException;
2320
import java.util.concurrent.TimeUnit;
21+
import java.util.concurrent.atomic.AtomicBoolean;
2422
import org.jetbrains.annotations.NotNull;
2523
import org.jetbrains.annotations.Nullable;
2624

@@ -35,9 +33,7 @@ public class MetricsBatchProcessor implements IMetricsBatchProcessor {
3533
private final @NotNull ISentryClient client;
3634
private final @NotNull Queue<SentryMetricsEvent> queue;
3735
private final @NotNull ISentryExecutorService executorService;
38-
private volatile @Nullable Future<?> scheduledFlush;
39-
private final @NotNull AutoClosableReentrantLock scheduleLock = new AutoClosableReentrantLock();
40-
private volatile boolean hasScheduled = false;
36+
private final @NotNull AtomicBoolean hasScheduled = new AtomicBoolean(false);
4137
private volatile boolean isShuttingDown = false;
4238

4339
private final @NotNull ReusableCountLatch pendingCount = new ReusableCountLatch();
@@ -69,15 +65,15 @@ public void add(final @NotNull SentryMetricsEvent metricsEvent) {
6965
}
7066
pendingCount.increment();
7167
queue.offer(metricsEvent);
72-
maybeSchedule(false, false);
68+
maybeSchedule(false);
7369
}
7470

7571
@SuppressWarnings("FutureReturnValueIgnored")
7672
@Override
7773
public void close(final boolean isRestarting) {
7874
isShuttingDown = true;
7975
if (isRestarting) {
80-
maybeSchedule(true, true);
76+
maybeSchedule(true);
8177
executorService.submit(() -> executorService.close(options.getShutdownTimeoutMillis()));
8278
} else {
8379
executorService.close(options.getShutdownTimeoutMillis());
@@ -87,33 +83,30 @@ public void close(final boolean isRestarting) {
8783
}
8884
}
8985

90-
private void maybeSchedule(boolean forceSchedule, boolean immediately) {
91-
if (hasScheduled && !forceSchedule) {
86+
@SuppressWarnings("FutureReturnValueIgnored")
87+
private void maybeSchedule(boolean immediately) {
88+
if (immediately) {
89+
// any already scheduled task may be far in the future, we want to schedule something that
90+
// runs right away
91+
hasScheduled.set(true);
92+
} else if (!hasScheduled.compareAndSet(false, true)) {
93+
// was already true, no need to schedule again
9294
return;
9395
}
94-
try (final @NotNull ISentryLifecycleToken ignored = scheduleLock.acquire()) {
95-
final @Nullable Future<?> latestScheduledFlush = scheduledFlush;
96-
if (forceSchedule
97-
|| latestScheduledFlush == null
98-
|| latestScheduledFlush.isDone()
99-
|| latestScheduledFlush.isCancelled()) {
100-
hasScheduled = true;
101-
final int flushAfterMs = immediately ? 0 : FLUSH_AFTER_MS;
102-
try {
103-
scheduledFlush = executorService.schedule(new BatchRunnable(), flushAfterMs);
104-
} catch (RejectedExecutionException e) {
105-
hasScheduled = false;
106-
options
107-
.getLogger()
108-
.log(SentryLevel.WARNING, "Metrics batch processor flush task rejected", e);
109-
}
110-
}
96+
final int flushAfterMs = immediately ? 0 : FLUSH_AFTER_MS;
97+
try {
98+
executorService.schedule(new BatchRunnable(), flushAfterMs);
99+
} catch (RejectedExecutionException e) {
100+
hasScheduled.set(false);
101+
options
102+
.getLogger()
103+
.log(SentryLevel.WARNING, "Metrics batch processor flush task rejected", e);
111104
}
112105
}
113106

114107
@Override
115108
public void flush(long timeoutMillis) {
116-
maybeSchedule(true, true);
109+
maybeSchedule(true);
117110
try {
118111
pendingCount.waitTillZero(timeoutMillis, TimeUnit.MILLISECONDS);
119112
} catch (InterruptedException e) {
@@ -124,12 +117,9 @@ public void flush(long timeoutMillis) {
124117

125118
private void flush() {
126119
flushInternal();
127-
try (final @NotNull ISentryLifecycleToken ignored = scheduleLock.acquire()) {
128-
if (!queue.isEmpty()) {
129-
maybeSchedule(true, false);
130-
} else {
131-
hasScheduled = false;
132-
}
120+
hasScheduled.set(false);
121+
if (!queue.isEmpty()) {
122+
maybeSchedule(false);
133123
}
134124
}
135125

‎sentry/src/test/java/io/sentry/logger/LoggerBatchProcessorTest.kt‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package io.sentry.logger
22

3+
import com.google.common.truth.Truth.assertThat
34
import io.sentry.DataCategory
45
import io.sentry.ISentryClient
56
import io.sentry.SentryLogEvent
@@ -21,9 +22,30 @@ import kotlin.test.assertTrue
2122
import org.mockito.kotlin.argumentCaptor
2223
import org.mockito.kotlin.atLeast
2324
import org.mockito.kotlin.mock
25+
import org.mockito.kotlin.times
2426
import org.mockito.kotlin.verify
2527

2628
class LoggerBatchProcessorTest {
29+
@Test
30+
fun `schedules another flush after previous flush has run`() {
31+
val mockClient = mock<ISentryClient>()
32+
val mockExecutor = DeferredExecutorService()
33+
val processor = LoggerBatchProcessor(SentryOptions(), mockClient, mockExecutor)
34+
35+
processor.add(SentryLogEvent(SentryId(), SentryNanotimeDate(), "first", SentryLogLevel.INFO))
36+
mockExecutor.runAll()
37+
38+
processor.add(SentryLogEvent(SentryId(), SentryNanotimeDate(), "second", SentryLogLevel.INFO))
39+
assertThat(mockExecutor.hasScheduledRunnables()).isTrue()
40+
mockExecutor.runAll()
41+
42+
val captor = argumentCaptor<SentryLogEvents>()
43+
verify(mockClient, times(2)).captureBatchedLogEvents(captor.capture())
44+
assertThat(captor.allValues.flatMap { it.items }.map { it.body })
45+
.containsExactly("first", "second")
46+
.inOrder()
47+
}
48+
2749
@Test
2850
fun `drops log events after reaching MAX_QUEUE_SIZE limit`() {
2951
// given

‎sentry/src/test/java/io/sentry/metrics/MetricsBatchProcessorTest.kt‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package io.sentry.metrics
22

3+
import com.google.common.truth.Truth.assertThat
34
import io.sentry.DataCategory
45
import io.sentry.ISentryClient
56
import io.sentry.SentryMetricsEvent
@@ -20,9 +21,31 @@ import kotlin.test.assertTrue
2021
import org.mockito.kotlin.argumentCaptor
2122
import org.mockito.kotlin.atLeast
2223
import org.mockito.kotlin.mock
24+
import org.mockito.kotlin.times
2325
import org.mockito.kotlin.verify
2426

2527
class MetricsBatchProcessorTest {
28+
@Test
29+
fun `schedules another flush after previous flush has run`() {
30+
val mockClient = mock<ISentryClient>()
31+
val mockExecutor = DeferredExecutorService()
32+
val processor = MetricsBatchProcessor(SentryOptions(), mockClient)
33+
processor.injectForField("executorService", mockExecutor)
34+
35+
processor.add(SentryMetricsEvent(SentryId(), SentryNanotimeDate(), "first", "gauge", 1.0))
36+
mockExecutor.runAll()
37+
38+
processor.add(SentryMetricsEvent(SentryId(), SentryNanotimeDate(), "second", "gauge", 2.0))
39+
assertThat(mockExecutor.hasScheduledRunnables()).isTrue()
40+
mockExecutor.runAll()
41+
42+
val captor = argumentCaptor<SentryMetricsEvents>()
43+
verify(mockClient, times(2)).captureBatchedMetricsEvents(captor.capture())
44+
assertThat(captor.allValues.flatMap { it.items }.map { it.name })
45+
.containsExactly("first", "second")
46+
.inOrder()
47+
}
48+
2649
@Test
2750
fun `drops metrics events after reaching MAX_QUEUE_SIZE limit`() {
2851
// given

0 commit comments

Comments
 (0)