From c677fef472d5ab325fbe273b8c859297135a18e7 Mon Sep 17 00:00:00 2001 From: "renovate[bot]" <29139614+renovate[bot]@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:15:23 +0000 Subject: [PATCH 1/4] deps: update dependency flagsmith/sdk-spec to v0.8.0 --- .github/workflows/sdk-spec.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/sdk-spec.yml b/.github/workflows/sdk-spec.yml index ec972bb..734eabb 100644 --- a/.github/workflows/sdk-spec.yml +++ b/.github/workflows/sdk-spec.yml @@ -11,8 +11,8 @@ permissions: jobs: sdk-spec: # Keep the ref in `uses:` and `spec-ref` in sync. - uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.7.0 + uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.8.0 with: - spec-ref: v0.7.0 + spec-ref: v0.8.0 language: java tags: core or local-evaluation From 12ba163f31ef4d6003517214aa0c3707e5df14b0 Mon Sep 17 00:00:00 2001 From: "flagsmith-engineering[bot]" <216534828+flagsmith-engineering[bot]@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:15:39 +0000 Subject: [PATCH 2/4] ci: Run the SDK spec for this SDK's capabilities Its entry in Flagsmith/sdk-spec's compatibility matrix at v0.8.0: `core or events or local-evaluation`. --- .github/workflows/sdk-spec.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/sdk-spec.yml b/.github/workflows/sdk-spec.yml index 734eabb..cce4a74 100644 --- a/.github/workflows/sdk-spec.yml +++ b/.github/workflows/sdk-spec.yml @@ -15,4 +15,4 @@ jobs: with: spec-ref: v0.8.0 language: java - tags: core or local-evaluation + tags: core or events or local-evaluation From 723a71291462cf9659112f218012c18034b8e833 Mon Sep 17 00:00:00 2001 From: "flagsmithsmith[bot]" <346993984+flagsmithsmith[bot]@users.noreply.github.com> Date: Thu, 29 Oct 2026 12:13:19 +0000 Subject: [PATCH 3/4] test(sdk-spec): Run the experimentation steps against the SDK Spec v0.7.1 runs the `@events` scenarios for this SDK: getting an experiment flag, tracking an event and tracking an exposure event, all with events enabled. The SDK already implements these (#226), so the steps file wires each step to FlagsmithClient.getExperimentFlag, trackEvent and trackExposureEvent, and creates the events-enabled client through FlagsmithConfig, next to the local evaluation client. (cherry picked from commit 0c48af8fcd55e8afd11b73947d351a02117e5ef7) Co-Authored-By: Claude Opus 5.5 (cherry picked from commit 29cbe7b9b49b9726359324188272600688c59836) --- sdk-spec/Steps.java | 43 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/sdk-spec/Steps.java b/sdk-spec/Steps.java index 74f05c6..01855c8 100644 --- a/sdk-spec/Steps.java +++ b/sdk-spec/Steps.java @@ -3,6 +3,7 @@ import com.flagsmith.exceptions.FlagsmithClientError; import com.flagsmith.models.BaseFlag; import com.flagsmith.models.Flags; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.stream.Collectors; @@ -47,6 +48,20 @@ static FlagsmithClient createClientWithLocalEvaluation(String environmentKey, St .build(); } + // Create the SDK client with events enabled + static FlagsmithClient createClientWithEvents( + String environmentKey, String apiUrl, String eventsUrl) { + return FlagsmithClient.newBuilder() + .setApiKey(environmentKey) + .withConfiguration( + FlagsmithConfig.newBuilder() + .baseUri(apiUrl) + .withEnableEvents(true) + .eventsUri(eventsUrl) + .build()) + .build(); + } + // When the application gets the environment flags static List getEnvironmentFlags(FlagsmithClient flagsmith) throws FlagsmithClientError { @@ -59,4 +74,32 @@ static List getIdentityFlags( throws FlagsmithClientError { return asSpecFlags(flagsmith.getIdentityFlags(identifier, traits)); } + + // When the application gets the experiment flag "{feature}" for identity "{identifier}" with traits: + static List getExperimentFlag( + FlagsmithClient flagsmith, String feature, String identifier, Map traits) + throws FlagsmithClientError { + return Collections.singletonList( + new SpecFlag(flagsmith.getExperimentFlag(feature, identifier, traits))); + } + + // When the application tracks event "{event}" for identity "{identifier}" with value {value} and metadata: + static void trackEvent( + FlagsmithClient flagsmith, + String event, + String identifier, + Object value, + Map metadata) { + flagsmith.trackEvent(event, identifier, value, null, metadata); + } + + // When the application tracks an exposure of feature "{feature}" for identity "{identifier}" with value {value} and metadata: + static void trackExposureEvent( + FlagsmithClient flagsmith, + String feature, + String identifier, + Object value, + Map metadata) { + flagsmith.trackExposureEvent(feature, identifier, value, null, metadata); + } } From 88fa258f47d83fe897b5b448b0126aca3fe4f789 Mon Sep 17 00:00:00 2001 From: "flagsmithsmith[bot]" <346993984+flagsmithsmith[bot]@users.noreply.github.com> Date: Wed, 11 Nov 2026 13:10:58 +0000 Subject: [PATCH 4/4] feat: Retry event delivery and close the client with a timeout Spec v0.8.0 specifies how SDKs deliver events. The event processor now: - Times out each events request after 3 s, and gives a batch 3 attempts, backing off from 1 s, doubling, with full jitter. Network errors, timeouts, 408, 429, 500, 502, 503 and 504 are retried. - Keeps a batch whose last attempt failed transiently in the buffer, ahead of newer events, for the next flush, unless that attempt got a 500. Any other error response drops the batch. - Stops event tracking, dropping the buffer, on a 401 or 403. - Sends one batch at a time, and caps the buffer at 1000 events by dropping the oldest, kept events included. - Never fails `flushEvents()`, which waits for the send, retries included. `FlagsmithClient.close(timeout, unit)` sends the buffered events for at most the timeout, then abandons the request in flight. `close()` waits for one batch's worst case, 12 s. The spec steps file wires the new steps: flushing events, closing with a timeout, tracking a range of events, and a client with both local evaluation and events enabled. Co-Authored-By: Claude Opus 5.5 --- sdk-spec/Steps.java | 33 ++ .../java/com/flagsmith/FlagsmithClient.java | 28 +- .../com/flagsmith/threads/EventProcessor.java | 468 ++++++++++++------ .../com/flagsmith/FlagsmithClientTest.java | 21 + .../flagsmith/threads/EventProcessorTest.java | 271 ++++++++-- 5 files changed, 621 insertions(+), 200 deletions(-) diff --git a/sdk-spec/Steps.java b/sdk-spec/Steps.java index 01855c8..3c5d9c3 100644 --- a/sdk-spec/Steps.java +++ b/sdk-spec/Steps.java @@ -6,6 +6,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; /** @@ -62,6 +63,21 @@ static FlagsmithClient createClientWithEvents( .build(); } + // Create the SDK client with local evaluation and events enabled + static FlagsmithClient createClientWithLocalEvaluationAndEvents( + String environmentKey, String apiUrl, String eventsUrl) { + return FlagsmithClient.newBuilder() + .setApiKey(environmentKey) + .withConfiguration( + FlagsmithConfig.newBuilder() + .baseUri(apiUrl) + .withLocalEvaluation(true) + .withEnableEvents(true) + .eventsUri(eventsUrl) + .build()) + .build(); + } + // When the application gets the environment flags static List getEnvironmentFlags(FlagsmithClient flagsmith) throws FlagsmithClientError { @@ -102,4 +118,21 @@ static void trackExposureEvent( Map metadata) { flagsmith.trackExposureEvent(feature, identifier, value, null, metadata); } + + // When the application tracks events "event_{first}" to "event_{last}" for identity "{identifier}" + static void trackEvents(FlagsmithClient flagsmith, int first, int last, String identifier) { + for (int number = first; number <= last; number++) { + flagsmith.trackEvent("event_" + number, identifier); + } + } + + // When the application flushes events + static void flushEvents(FlagsmithClient flagsmith) throws Exception { + flagsmith.flushEvents().get(); + } + + // When the application closes the client with a timeout of {timeout} seconds + static void close(FlagsmithClient flagsmith, int timeout) { + flagsmith.close(timeout, TimeUnit.SECONDS); + } } diff --git a/src/main/java/com/flagsmith/FlagsmithClient.java b/src/main/java/com/flagsmith/FlagsmithClient.java index 185979e..c01b3ea 100644 --- a/src/main/java/com/flagsmith/FlagsmithClient.java +++ b/src/main/java/com/flagsmith/FlagsmithClient.java @@ -31,6 +31,8 @@ import java.util.Map; import java.util.Objects; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import java.util.function.Function; import java.util.stream.Collectors; import lombok.AccessLevel; @@ -408,8 +410,9 @@ public void trackExposureEvent(String featureName, String identifier, Object val /** * Send buffered events now. * - * @return a future completing once every event buffered so far has been sent or dropped, already - * completed when events are not enabled + * @return a future completing, never exceptionally, once every event buffered so far has been + * sent, dropped or kept for the next flush, retries included; already completed when events + * are not enabled */ public CompletableFuture flushEvents() { if (eventProcessor == null) { @@ -421,15 +424,32 @@ public CompletableFuture flushEvents() { /** * Should be called when terminating the client to clean up any resources that - * need cleaning up. + * need cleaning up. Sends the buffered events first, waiting for at most one batch's worst case + * under the events API's timeouts and retries. **/ public void close() { + close(EventProcessor::close); + } + + /** + * Should be called when terminating the client to clean up any resources that need cleaning + * up. Sends the buffered events first, retries included, waiting for them for at most + * {@code timeout}; events still unsent then are dropped. Never throws. + * + * @param timeout how long to wait for the buffered events to be sent + * @param unit unit of {@code timeout} + */ + public void close(long timeout, TimeUnit unit) { + close((processor) -> processor.close(unit.toMillis(timeout))); + } + + private void close(Consumer closeEvents) { if (pollingManager != null) { pollingManager.stopPolling(); } if (eventProcessor != null) { - eventProcessor.close(); + closeEvents.accept(eventProcessor); } flagsmithSdk.close(); diff --git a/src/main/java/com/flagsmith/threads/EventProcessor.java b/src/main/java/com/flagsmith/threads/EventProcessor.java index 0da9e9d..767d88e 100644 --- a/src/main/java/com/flagsmith/threads/EventProcessor.java +++ b/src/main/java/com/flagsmith/threads/EventProcessor.java @@ -1,16 +1,15 @@ package com.flagsmith.threads; import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.util.RawValue; import com.flagsmith.FlagsmithLogger; import com.flagsmith.MapperFactory; import com.flagsmith.Versions; -import com.flagsmith.config.Retry; import com.flagsmith.exceptions.FlagsmithRuntimeError; import com.flagsmith.interfaces.FlagsmithSdk; import com.flagsmith.models.TraitConfig; +import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -22,27 +21,37 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.stream.Collectors; -import java.util.stream.IntStream; +import java.util.function.LongUnaryOperator; import lombok.AccessLevel; import lombok.Getter; import lombok.Setter; +import okhttp3.Call; import okhttp3.HttpUrl; import okhttp3.MediaType; import okhttp3.OkHttpClient; import okhttp3.Request; import okhttp3.RequestBody; +import okhttp3.Response; +import okhttp3.ResponseBody; /** * Buffers experimentation events and sends them to the Flagsmith events API, on a timer, when the * buffer fills, and on {@link #close()}. Exposures are deduplicated within a flush window. + * + *

One batch is in flight at a time. Each request times out after + * {@link #REQUEST_TIMEOUT_MILLIS}, and a batch gets {@link #MAX_ATTEMPTS} attempts, backing off + * from {@link #INITIAL_BACKOFF_MILLIS}, doubling, with full jitter. Transient failures (network + * errors, timeouts, 408, 429, 500, 502, 503 and 504) are retried; a batch whose last attempt still + * failed transiently is kept, ahead of newer events, for the next flush, unless that attempt got a + * 500. Any other error response drops the batch, and a 401 or 403 stops event tracking altogether. */ public class EventProcessor { @@ -59,11 +68,38 @@ public class EventProcessor { private static final String EXPERIMENT_ID_KEY = "experiment_id"; private static final MediaType JSON_MEDIA_TYPE = MediaType.get("application/json; charset=utf-8"); - static final int MAX_IN_FLIGHT_BATCHES = 2; static final int MAX_BUFFERED_EVENTS = 1_000; - static final long CLOSE_TIMEOUT_MILLIS = 25_000L; + static final int MAX_ATTEMPTS = 3; + static final long REQUEST_TIMEOUT_MILLIS = 3_000L; + static final long INITIAL_BACKOFF_MILLIS = 1_000L; + /** Every attempt of one batch timing out, with the longest backoff before each retry. */ + static final long CLOSE_TIMEOUT_MILLIS = MAX_ATTEMPTS * REQUEST_TIMEOUT_MILLIS + + INITIAL_BACKOFF_MILLIS * ((1L << (MAX_ATTEMPTS - 1)) - 1); private static final long DROP_LOG_INTERVAL_NANOS = TimeUnit.SECONDS.toNanos(10); + /** How a batch's delivery ended. */ + enum Outcome { + /** The events API accepted it. */ + SENT, + /** Its last attempt failed transiently: buffer it again for the next flush. */ + KEPT, + /** It is lost. */ + DROPPED, + /** The events API refused the environment key: stop tracking events. */ + UNAUTHORISED + } + + /** How one attempt to send a batch ended. */ + private enum Attempt { + ACCEPTED, + /** A network error, a timeout or a status worth retrying. */ + UNAVAILABLE, + /** A 500: retried, but the batch is dropped if the last attempt gets one. */ + SERVER_ERROR, + REJECTED, + UNAUTHORISED + } + @Getter private final HttpUrl eventsEndpoint; @Getter @@ -76,26 +112,33 @@ public class EventProcessor { private final Object lock = new Object(); @Getter(AccessLevel.PACKAGE) private final ScheduledExecutorService scheduler; - private final Set> inFlight = ConcurrentHashMap.newKeySet(); - private CompletableFuture nextBatch = new CompletableFuture<>(); - private int droppedSinceLastReport = 0; // guarded by lock - private Long lastDropReportNanos = null; // guarded by lock @Getter(AccessLevel.PACKAGE) - private final RequestProcessor requestProcessor; + private final ExecutorService sender; + @Getter(AccessLevel.PACKAGE) + private final OkHttpClient client; + private final LongUnaryOperator jitter; + private CompletableFuture inFlight = null; // guarded by lock + private CompletableFuture nextBatch = null; // guarded by lock + private int droppedSinceLastReport = 0; // guarded by lock + private Long lastDropReportNanos = null; // guarded by lock + private volatile Call currentCall; + private volatile boolean cancelled = false; @Getter(AccessLevel.PACKAGE) @Setter(AccessLevel.PACKAGE) - private long closeTimeoutMillis; + private long closeTimeoutMillis = CLOSE_TIMEOUT_MILLIS; @Setter private FlagsmithSdk api; private FlagsmithLogger logger = new FlagsmithLogger(); private final AtomicBoolean closed = new AtomicBoolean(false); + private final AtomicBoolean disabled = new AtomicBoolean(false); private final AtomicBoolean claimed = new AtomicBoolean(false); private ScheduledFuture scheduledFlush; /** * Create a processor that sends batches through {@code client}. * - * @param client HTTP client; its timeouts also bound {@link #close()} + * @param client HTTP client; its interceptors apply, but each events request + * times out after {@link #REQUEST_TIMEOUT_MILLIS} * @param eventsUri base URI of the events API, e.g. https://events.api.flagsmith.com/ * @param maxBufferItems number of buffered events that triggers an immediate flush; at * least 1 @@ -106,11 +149,16 @@ public class EventProcessor { public EventProcessor(OkHttpClient client, HttpUrl eventsUri, int maxBufferItems, int flushIntervalMillis) { this(eventsUri, maxBufferItems, flushIntervalMillis, - new RequestProcessor(withCallDeadline(client), new FlagsmithLogger(), buildRetry())); + client.newBuilder() + .callTimeout(REQUEST_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS) + // Every attempt is one request, retried below with a backoff. + .retryOnConnectionFailure(false) + .build(), + EventProcessor::fullJitter); } EventProcessor(HttpUrl eventsUri, int maxBufferItems, int flushIntervalMillis, - RequestProcessor requestProcessor) { + OkHttpClient client, LongUnaryOperator jitter) { if (maxBufferItems < 1) { throw new IllegalArgumentException("maxBufferItems must be at least 1."); } @@ -120,55 +168,23 @@ public EventProcessor(OkHttpClient client, HttpUrl eventsUri, int maxBufferItems this.eventsEndpoint = eventsUri.newBuilder(EVENTS_PATH).build(); this.maxBufferItems = maxBufferItems; this.flushIntervalMillis = flushIntervalMillis; - this.requestProcessor = requestProcessor; - this.closeTimeoutMillis = worstCaseBatchMillis(requestProcessor.getClient(), buildRetry()); + this.client = client; + this.jitter = jitter; this.scheduler = Executors.newSingleThreadScheduledExecutor((runnable) -> { Thread thread = new Thread(runnable, "flagsmith-events"); thread.setDaemon(true); return thread; }); + this.sender = Executors.newSingleThreadExecutor((runnable) -> { + Thread thread = new Thread(runnable, "flagsmith-events-sender"); + thread.setDaemon(true); + return thread; + }); } - private static Retry buildRetry() { - Retry retry = new Retry(2); - retry.setStatusForcelist( - IntStream.rangeClosed(500, 599).boxed().collect(Collectors.toSet())); - retry.setStatusForcelistOnly(Boolean.TRUE); - return retry; - } - - private static long worstCaseBatchMillis(OkHttpClient client, Retry retry) { - long attemptMillis = attemptMillis(client); - if (attemptMillis == 0) { - return CLOSE_TIMEOUT_MILLIS; - } - long total = retry.getTotal() * attemptMillis; - for (int retried = 1; retried < retry.getTotal(); retried++) { - total += (long) (Math.min(retry.getBackoffFactor() * 2 * retried, retry.getBackoffMax()) - * 1000); - } - return total; - } - - private static long attemptMillis(OkHttpClient client) { - if (client.callTimeoutMillis() > 0) { - return client.callTimeoutMillis(); - } - if (client.connectTimeoutMillis() > 0 && client.writeTimeoutMillis() > 0 - && client.readTimeoutMillis() > 0) { - return (long) client.connectTimeoutMillis() + client.writeTimeoutMillis() - + client.readTimeoutMillis(); - } - return 0; - } - - private static OkHttpClient withCallDeadline(OkHttpClient client) { - if (client.callTimeoutMillis() > 0) { - return client; - } - long attemptMillis = attemptMillis(client); - long callTimeout = attemptMillis > 0 ? attemptMillis : CLOSE_TIMEOUT_MILLIS; - return client.newBuilder().callTimeout(callTimeout, TimeUnit.MILLISECONDS).build(); + /** A random wait from none up to {@code backoffMillis}. */ + static long fullJitter(long backoffMillis) { + return ThreadLocalRandom.current().nextLong(backoffMillis + 1); } /** @@ -183,13 +199,12 @@ public void claim() { } /** - * Set the logger, for this processor and its request processor. + * Set the logger. * * @param logger the client's logger, so event failures appear alongside its other output */ public void setLogger(FlagsmithLogger logger) { this.logger = logger; - requestProcessor.setLogger(logger); } /** @@ -222,35 +237,37 @@ public void trackExposureEvent(String featureName, String identifier, Object val } /** - * Send everything buffered so far. + * Send everything buffered so far. While a batch is in flight, the buffer is sent once it + * succeeds or is dropped; if it is kept instead, the buffer waits for the next flush with it. * - * @return a future completing once every event buffered so far has been sent or dropped + * @return a future completing, never exceptionally, once every event buffered so far has been + * sent, dropped or kept for the next flush, retries included */ public CompletableFuture flush() { - List> batch = null; - CompletableFuture tracked = null; - CompletableFuture waiting = null; + List> batch; + CompletableFuture tracked; synchronized (lock) { - if (!buffer.isEmpty() && (inFlight.size() < MAX_IN_FLIGHT_BATCHES || closed.get())) { - batch = new ArrayList<>(buffer); - buffer.clear(); - dedupeKeys.clear(); - dedupeKeyByEvent.clear(); - tracked = nextBatch; - nextBatch = new CompletableFuture<>(); - inFlight.add(tracked); - } else if (!buffer.isEmpty()) { - waiting = nextBatch; + if (disabled.get()) { + clearBuffer(); + return CompletableFuture.completedFuture(null); } + if (inFlight != null) { + if (nextBatch == null && !buffer.isEmpty()) { + nextBatch = new CompletableFuture<>(); + } + return nextBatch == null ? inFlight : nextBatch; + } + if (buffer.isEmpty()) { + return CompletableFuture.completedFuture(null); + } + batch = takeBuffer(); + tracked = new CompletableFuture<>(); + inFlight = tracked; } - if (batch != null) { - send(batch, tracked); - } - - CompletableFuture sent = awaitInFlight(); - return waiting == null ? sent : CompletableFuture.allOf(sent, waiting); + send(batch, tracked); + return tracked; } /** @@ -272,29 +289,52 @@ public synchronized void start() { } /** - * Stop the timer, flush what is left and release HTTP resources. Blocks until in-flight - * batches settle, for at most one batch's worst case under the client's timeouts, or - * {@link #CLOSE_TIMEOUT_MILLIS} if a timeout is off. + * Stop the timer and send what is left, waiting for at most one batch's worst case: + * {@link #CLOSE_TIMEOUT_MILLIS}. */ public void close() { + close(closeTimeoutMillis); + } + + /** + * Stop the timer and send what is left, retries included, for at most {@code timeoutMillis}. + * Then abandon the request in flight and drop the events still unsent. Never throws. + * + * @param timeoutMillis how long to wait for the events to be sent + */ + public void close(long timeoutMillis) { synchronized (this) { closed.set(true); scheduler.shutdownNow(); } try { - flush().get(closeTimeoutMillis, TimeUnit.MILLISECONDS); + flush().get(Math.max(timeoutMillis, 0), TimeUnit.MILLISECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); logger.error("Interrupted while flushing events on close.", e); } catch (TimeoutException e) { - logger.error("Stopped waiting for events to be delivered after " + closeTimeoutMillis + logger.error("Stopped waiting for events to be delivered after " + timeoutMillis + "ms on close."); } catch (Exception e) { logger.error("Failed to flush events on close.", e); } - requestProcessor.close(); + cancelled = true; + sender.shutdownNow(); + Call call = currentCall; + if (call != null) { + call.cancel(); + } + + int unsent; + synchronized (lock) { + unsent = buffer.size(); + clearBuffer(); + } + if (unsent > 0) { + logger.error("Dropped " + unsent + " events left unsent on close."); + } } private void bufferEvent(String event, String featureName, String identifier, Object value, @@ -303,6 +343,9 @@ private void bufferEvent(String event, String featureName, String identifier, Ob logClosed(event); return; } + if (disabled.get()) { + return; + } try { final String stringValue = value == null ? null : String.valueOf(value); @@ -325,14 +368,17 @@ private void bufferEvent(String event, String featureName, String identifier, Ob eventPayload.put("metadata", toJson(eventMetadata)); eventPayload.put("timestamp", System.currentTimeMillis()); - boolean isFull = false; - int droppedToReport = 0; + boolean isFull; + int droppedToReport; synchronized (lock) { if (closed.get()) { logClosed(event); return; } + if (disabled.get()) { + return; + } if (dedupe) { List key = dedupeKey(event, featureName, identifier, stringValue, experimentId); if (!dedupeKeys.add(key)) { @@ -341,12 +387,8 @@ private void bufferEvent(String event, String featureName, String identifier, Ob dedupeKeyByEvent.put(eventPayload, key); } buffer.add(eventPayload); - if (inFlight.size() < MAX_IN_FLIGHT_BATCHES) { - isFull = buffer.size() >= maxBufferItems; - } else if (buffer.size() > Math.max(maxBufferItems, MAX_BUFFERED_EVENTS)) { - dedupeKeys.remove(dedupeKeyByEvent.remove(buffer.remove(0))); - droppedToReport = recordDrop(1); - } + droppedToReport = dropOverflow(); + isFull = buffer.size() >= maxBufferItems; } logDrops(droppedToReport); @@ -389,66 +431,205 @@ private static List dedupeKey(String event, String featureName, String i experimentId == null ? null : String.valueOf(experimentId)); } - private void send(List> batch, CompletableFuture tracked) { - final int batchSize = batch.size(); - boolean submitted = false; + /** Empties the buffer into a batch, opening a new dedupe window. Call holding the lock. */ + private List> takeBuffer() { + List> batch = new ArrayList<>(buffer); + clearBuffer(); + return batch; + } + /** Call holding the lock. */ + private void clearBuffer() { + buffer.clear(); + dedupeKeys.clear(); + dedupeKeyByEvent.clear(); + } + + /** + * Drops the oldest events beyond the buffer's capacity. Call holding the lock. + * + * @return the number of dropped events to log now + */ + private int dropOverflow() { + int overflow = buffer.size() - Math.max(maxBufferItems, MAX_BUFFERED_EVENTS); + if (overflow <= 0) { + return 0; + } + List> oldest = buffer.subList(0, overflow); + for (Map event : oldest) { + dedupeKeys.remove(dedupeKeyByEvent.remove(event)); + } + oldest.clear(); + return recordDrop(overflow); + } + + private void send(List> batch, CompletableFuture tracked) { try { - if (api == null) { - logger.error("Dropping " + batchSize - + " events: the event processor has no API wrapper."); - return; + Request request = buildRequest(batch); + sender.execute(() -> { + Outcome delivered = Outcome.DROPPED; + try { + delivered = deliver(request, batch.size()); + } catch (RuntimeException e) { + logger.error("Dropping " + batch.size() + " events: failed to send them.", e); + } finally { + settle(batch, delivered, tracked); + } + }); + return; + } catch (Exception e) { + logger.error("Dropping " + batch.size() + " events: failed to send them.", e); + } + settle(batch, Outcome.DROPPED, tracked); + } + + private Request buildRequest(List> batch) throws JsonProcessingException { + if (api == null) { + throw new IllegalStateException("The event processor has no API wrapper."); + } + + String payload = MapperFactory.getMapper() + .writeValueAsString(Collections.singletonMap("events", batch)); + + RequestBody body = RequestBody.create(payload, JSON_MEDIA_TYPE); + return new Request.Builder() + .url(eventsEndpoint) + .post(body) + .header(AUTH_HEADER, api.newPostRequest(eventsEndpoint, body).headers(AUTH_HEADER).get(0)) + .header(USER_AGENT_HEADER, SDK_USER_AGENT_PREFIX + Versions.getVersion()) + .header(SDK_USER_AGENT_HEADER, SDK_USER_AGENT_PREFIX + Versions.getVersion()) + .header(ACCEPT_HEADER, "application/json") + .build(); + } + + /** Sends one batch, retrying transient failures. Runs on the sender thread. */ + private Outcome deliver(Request request, int batchSize) { + for (int attempt = 1; ; attempt++) { + if (attempt > 1) { + try { + Thread.sleep(jitter.applyAsLong(INITIAL_BACKOFF_MILLIS << (attempt - 2))); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return Outcome.DROPPED; + } + } + if (cancelled) { + return Outcome.DROPPED; } - String payload = MapperFactory.getMapper() - .writeValueAsString(Collections.singletonMap("events", batch)); - - RequestBody body = RequestBody.create(payload, JSON_MEDIA_TYPE); - Request request = new Request.Builder() - .url(eventsEndpoint) - .post(body) - .header(AUTH_HEADER, api.newPostRequest(eventsEndpoint, body).headers(AUTH_HEADER).get(0)) - .header(USER_AGENT_HEADER, SDK_USER_AGENT_PREFIX + Versions.getVersion()) - .header(SDK_USER_AGENT_HEADER, SDK_USER_AGENT_PREFIX + Versions.getVersion()) - .header(ACCEPT_HEADER, "application/json") - .build(); - - requestProcessor - .submit(request, new TypeReference() {}, Boolean.FALSE, buildRetry()) - .whenComplete((response, error) -> { - try { - logRejections(response, batchSize); - } finally { - settle(tracked); - } - }); - submitted = true; - } catch (Exception e) { - logger.error("Dropping " + batchSize + " events: failed to send them.", e); - } finally { - if (!submitted) { - settle(tracked); + switch (attempt(request, batchSize)) { + case ACCEPTED: + return Outcome.SENT; + case REJECTED: + return Outcome.DROPPED; + case UNAUTHORISED: + return Outcome.UNAUTHORISED; + case SERVER_ERROR: + if (attempt == MAX_ATTEMPTS) { + logger.error("Dropping " + batchSize + " events: the events API failed to take them " + + MAX_ATTEMPTS + " times."); + return Outcome.DROPPED; + } + break; + default: + if (attempt == MAX_ATTEMPTS) { + logger.error("Could not reach the events API after " + MAX_ATTEMPTS + " attempts. " + + "Keeping " + batchSize + " events for the next flush."); + return Outcome.KEPT; + } } } } - private void logRejections(JsonNode response, int batchSize) { - JsonNode rejected = response == null ? null : response.get("rejected"); + private Attempt attempt(Request request, int batchSize) { + Call call = client.newCall(request); + currentCall = call; + if (cancelled) { + call.cancel(); + } + + try (Response response = call.execute()) { + int status = response.code(); + if (response.isSuccessful()) { + logRejections(response.body(), batchSize); + return Attempt.ACCEPTED; + } + switch (status) { + case 401: + case 403: + return Attempt.UNAUTHORISED; + case 500: + logger.info("The events API answered {} to {} events.", status, batchSize); + return Attempt.SERVER_ERROR; + case 408: + case 429: + case 502: + case 503: + case 504: + logger.info("The events API answered {} to {} events.", status, batchSize); + return Attempt.UNAVAILABLE; + default: + logger.error("Dropping " + batchSize + " events: the events API rejected them with " + + status + "."); + return Attempt.REJECTED; + } + } catch (IOException e) { + logger.info("Failed to send {} events to the events API: {}", batchSize, e.toString()); + return Attempt.UNAVAILABLE; + } finally { + currentCall = null; + } + } + + private void logRejections(ResponseBody body, int batchSize) { + JsonNode rejected; + try { + rejected = body == null ? null : MapperFactory.getMapper().readTree(body.string()) + .get("rejected"); + } catch (IOException | RuntimeException e) { + return; + } if (rejected != null && rejected.isArray() && rejected.size() > 0) { logger.error("The events API rejected " + rejected.size() + " of " + batchSize + " events. First rejection: " + rejected.get(0)); } } - private void settle(CompletableFuture tracked) { - boolean waiting; + private void settle(List> batch, Outcome outcome, + CompletableFuture tracked) { + List> next = null; + CompletableFuture waiting; + int droppedToReport = 0; + synchronized (lock) { - inFlight.remove(tracked); - waiting = !buffer.isEmpty(); + if (outcome == Outcome.UNAUTHORISED && disabled.compareAndSet(false, true)) { + logger.error("The events API refused the environment key. Event tracking is stopped and " + + (batch.size() + buffer.size()) + " events are dropped."); + } + if (disabled.get()) { + clearBuffer(); + } else if (outcome == Outcome.KEPT && !closed.get()) { + buffer.addAll(0, batch); + droppedToReport = dropOverflow(); + } + + waiting = nextBatch; + nextBatch = null; + // A kept batch means the events API is failing: the buffer waits for the next flush. + if (waiting != null && outcome != Outcome.KEPT && !buffer.isEmpty()) { + next = takeBuffer(); + inFlight = waiting; + } else { + inFlight = null; + } } + + logDrops(droppedToReport); tracked.complete(null); - if (waiting) { - flush(); + if (next != null) { + send(next, waiting); + } else if (waiting != null) { + waiting.complete(null); } } @@ -466,17 +647,12 @@ private int recordDrop(int count) { private void logDrops(int dropped) { if (dropped > 0) { - logger.error("Dropped the " + dropped + " oldest events: " + MAX_IN_FLIGHT_BATCHES - + " batches are in flight to the events API and the buffer is full. Further drops are" - + " reported at most every " + TimeUnit.NANOSECONDS.toSeconds(DROP_LOG_INTERVAL_NANOS) - + "s."); + logger.error("Dropped the " + dropped + " oldest events: the buffer holds " + + Math.max(maxBufferItems, MAX_BUFFERED_EVENTS) + " events. Further drops are reported" + + " at most every " + TimeUnit.NANOSECONDS.toSeconds(DROP_LOG_INTERVAL_NANOS) + "s."); } } - private CompletableFuture awaitInFlight() { - return CompletableFuture.allOf(inFlight.toArray(new CompletableFuture[0])); - } - List> bufferedEvents() { synchronized (lock) { return new ArrayList<>(buffer); diff --git a/src/test/java/com/flagsmith/FlagsmithClientTest.java b/src/test/java/com/flagsmith/FlagsmithClientTest.java index e9625f1..6fb7c63 100644 --- a/src/test/java/com/flagsmith/FlagsmithClientTest.java +++ b/src/test/java/com/flagsmith/FlagsmithClientTest.java @@ -1101,6 +1101,27 @@ public void testEventApisRejectReservedAndBlankNames(Consumer c verify(processor, never()).trackExposureEvent(any(), any(), any(), any(), any()); } + @Test + public void testCloseWithATimeoutClosesTheEventProcessorWithIt() { + EventProcessor processor = mock(EventProcessor.class); + FlagsmithClient client = experimentClient(processor); + + client.close(2, TimeUnit.SECONDS); + + verify(processor).close(2_000L); + verify(processor, never()).close(); + } + + @Test + public void testCloseClosesTheEventProcessorWithItsDefaultTimeout() { + EventProcessor processor = mock(EventProcessor.class); + FlagsmithClient client = experimentClient(processor); + + client.close(); + + verify(processor).close(); + } + @Test public void testTrackExposureEventWithBlankIdentifierSendsNothing() { EventProcessor processor = mock(EventProcessor.class); diff --git a/src/test/java/com/flagsmith/threads/EventProcessorTest.java b/src/test/java/com/flagsmith/threads/EventProcessorTest.java index 4276a60..936486a 100644 --- a/src/test/java/com/flagsmith/threads/EventProcessorTest.java +++ b/src/test/java/com/flagsmith/threads/EventProcessorTest.java @@ -70,6 +70,7 @@ public class EventProcessorTest { private MockInterceptor interceptor; private RecordingInterceptor recorder; + private List backoffs; private FlagsmithSdk api; private EventProcessor eventProcessor; @@ -77,6 +78,7 @@ public class EventProcessorTest { public void init() { interceptor = new MockInterceptor(); recorder = new RecordingInterceptor(); + backoffs = Collections.synchronizedList(new ArrayList<>()); api = mock(FlagsmithSdk.class); when(api.newPostRequest(any(), any())).thenAnswer((invocation) -> new Request.Builder() .url((HttpUrl) invocation.getArgument(0)) @@ -108,11 +110,16 @@ private EventProcessor newProcessor( clientBuilder.addInterceptor(extra); } OkHttpClient client = clientBuilder.addInterceptor(interceptor).build(); + // Records each backoff and retries at once, so retry tests do not wait for real backoffs. eventProcessor = new EventProcessor( HttpUrl.get(EVENTS_URI), maxBufferItems, flushIntervalMillis, - new RequestProcessor(client, new FlagsmithLogger())); + client, + (backoff) -> { + backoffs.add(backoff); + return 0; + }); eventProcessor.setApi(api); return eventProcessor; } @@ -283,11 +290,11 @@ public void flush_completesOnlyAfterTheInFlightPostCompletes() { */ @ParameterizedTest @CsvSource({ - "503, 1, 2", "501, 1, 2", "500, 2, 2", - "400, 1, 1", - ", 1, 2", ", 2, 2"}) + "503, 2, 3", "502, 1, 2", "504, 1, 2", "408, 1, 2", "429, 1, 2", "500, 2, 3", + "400, 1, 1", "404, 1, 1", "501, 1, 1", + ", 1, 2", ", 2, 3"}) @SneakyThrows - public void flush_retriesOnceOnAServerErrorOrConnectionFailure( + public void flush_retriesTransientFailuresAndDropsRejectedBatches( Integer status, int failures, int expectedAttempts) { EventProcessor processor = status == null ? newProcessor(1000, 0, new FailingInterceptor(failures)) @@ -305,6 +312,129 @@ public void flush_retriesOnceOnAServerErrorOrConnectionFailure( assertTrue(processor.bufferedEvents().isEmpty()); } + @Test + public void flush_backsOffFromOneSecondDoublingBeforeEachRetry() { + EventProcessor processor = newProcessor(1000, 0); + interceptor.addRule().post(EVENTS_ENDPOINT).times(2).respond(503); + interceptor.addRule().post(EVENTS_ENDPOINT).anyTimes().respond(ACCEPTED_BODY, MEDIATYPE_JSON); + + processor.trackEvent("purchase", "user-1", "1", null, null); + flushAndWait(processor); + + assertEquals(3, recorder.count()); + assertEquals(Arrays.asList(1_000L, 2_000L), backoffs); + } + + @Test + public void fullJitter_waitsAnythingFromNoneUpToTheBackoff() { + for (int i = 0; i < 1000; i++) { + long wait = EventProcessor.fullJitter(1_000L); + assertTrue(wait >= 0 && wait <= 1_000L, "waited " + wait + "ms"); + } + assertEquals(0L, EventProcessor.fullJitter(0L)); + } + + /** + * Every attempt fails with the given responses, a status code or, when empty, a connection + * failure. The last one decides: a 500 drops the batch, any other transient failure keeps it. + */ + @ParameterizedTest + @CsvSource({"503, 503, 503, true", "500, 500, 503, true", "503, 503, 500, false", + ", , , true", "500, 500, , true"}) + @SneakyThrows + public void flush_keepsABatchForTheNextFlushUnlessItsLastAttemptGetsA500( + Integer first, Integer second, Integer last, boolean kept) { + EventProcessor processor = newProcessor(1000, 0, failingWith(first, second, last)); + interceptor.addRule().post(EVENTS_ENDPOINT).anyTimes().respond(ACCEPTED_BODY, MEDIATYPE_JSON); + + processor.trackEvent("add_to_cart", "user-1", "1", null, null); + flushAndWait(processor); + + assertEquals(EventProcessor.MAX_ATTEMPTS, recorder.count()); + assertEquals(kept ? 1 : 0, processor.bufferedEvents().size()); + + processor.trackEvent("checkout", "user-1", "42", null, null); + flushAndWait(processor); + + assertEquals(EventProcessor.MAX_ATTEMPTS + 1, recorder.count()); + JsonNode events = MapperFactory.getMapper() + .readTree(recorder.bodies().get(EventProcessor.MAX_ATTEMPTS)).get("events"); + List sent = new ArrayList<>(); + events.forEach((event) -> sent.add(event.get("event").asText())); + assertEquals(kept ? Arrays.asList("add_to_cart", "checkout") + : Collections.singletonList("checkout"), sent); + assertTrue(processor.bufferedEvents().isEmpty()); + } + + /** Answers each request with the next status, or fails its connection on null. */ + private static Interceptor failingWith(Integer... responses) { + AtomicInteger next = new AtomicInteger(); + return (chain) -> { + int index = next.getAndIncrement(); + if (index >= responses.length) { + return chain.proceed(chain.request()); + } + if (responses[index] == null) { + throw new IOException("connection reset"); + } + return new Response.Builder() + .request(chain.request()) + .protocol(Protocol.HTTP_1_1) + .code(responses[index]) + .message("Failed") + .body(ResponseBody.create("{}", MediaType.get("application/json"))) + .build(); + }; + } + + @ParameterizedTest + @CsvSource({"401", "403"}) + public void flush_stopsEventTrackingWhenTheEventsApiRefusesTheKey(int status) { + EventProcessor processor = newProcessor(1000, 0); + FlagsmithLogger logger = mockLogger(processor); + interceptor.addRule().post(EVENTS_ENDPOINT).times(1).respond(status); + interceptor.addRule().post(EVENTS_ENDPOINT).anyTimes().respond(ACCEPTED_BODY, MEDIATYPE_JSON); + + processor.trackEvent("add_to_cart", "user-1", "1", null, null); + flushAndWait(processor); + processor.trackEvent("checkout", "user-1", "42", null, null); + processor.trackExposureEvent("checkout_cta", "user-1", "treatment", null, null); + flushAndWait(processor); + + assertEquals(1, recorder.count()); + assertTrue(processor.bufferedEvents().isEmpty()); + verify(logger, times(1)).error(contains("refused the environment key")); + } + + @Test + public void flush_doesNotResendAKeptBatchForAFlushRequestedWhileItWasInFlight() { + CountDownLatch release = new CountDownLatch(1); + EventProcessor processor = newProcessor(1000, 0, (chain) -> { + try { + release.await(WAIT_SECONDS, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + throw new IOException("connection reset"); + }); + mockLogger(processor); + + processor.trackEvent("add_to_cart", "user-1", "1", null, null); + CompletableFuture first = processor.flush(); + processor.trackEvent("checkout", "user-1", "42", null, null); + CompletableFuture second = processor.flush(); + release.countDown(); + + first.join(); + second.join(); + assertEquals(EventProcessor.MAX_ATTEMPTS, recorder.count()); + assertEquals(2, processor.bufferedEvents().size()); + assertEquals("add_to_cart", processor.bufferedEvents().get(0).get("event")); + eventProcessor = null; + processor.close(0); + } + @Test @SneakyThrows public void flush_waitsForABatchAnotherThreadIsAlreadySending() { @@ -344,7 +474,7 @@ private static Stream> brokenSends() { return Stream.of( (test) -> doThrow(new IllegalStateException("boom")).when(test.api) .newPostRequest(any(), any()), - (test) -> test.eventProcessor.getRequestProcessor().close(), + (test) -> test.eventProcessor.getSender().shutdown(), (test) -> test.eventProcessor.setApi(null)); } @@ -529,50 +659,38 @@ public void start_isANoOpAfterClose() { assertTrue(processor.getScheduler().isShutdown()); } - private static Stream clientTimeouts() { - FlagsmithConfig longRead = FlagsmithConfig.newBuilder().readTimeout(30_000).build(); - return Stream.of( - // Two attempts at 2s connect + 5s write + 5s read, and 200ms backoff before the second. - Arguments.of(FlagsmithConfig.newBuilder().build().getHttpClient(), 2 * 12_000 + 200, - 12_000), - Arguments.of(longRead.getHttpClient(), 2 * 37_000 + 200, 37_000), - Arguments.of(new OkHttpClient.Builder().callTimeout(4, TimeUnit.SECONDS).build(), - 2 * 4_000 + 200, 4_000), - Arguments.of(new OkHttpClient.Builder().readTimeout(0, TimeUnit.SECONDS).build(), - 2 * EventProcessor.CLOSE_TIMEOUT_MILLIS + 200, EventProcessor.CLOSE_TIMEOUT_MILLIS)); - } - - @ParameterizedTest - @MethodSource("clientTimeouts") - public void close_waitsForOneBatchAtTheClientTimeouts(OkHttpClient client, long expected, - long callTimeout) { + @Test + public void eventProcessor_timesOutEachRequestAfterThreeSecondsAndRetriesItself() { + OkHttpClient client = FlagsmithConfig.newBuilder().readTimeout(30_000).build().getHttpClient(); EventProcessor processor = new EventProcessor(client, HttpUrl.get(EVENTS_URI), 1, 0); - assertEquals(expected, processor.getCloseTimeoutMillis()); - assertEquals(callTimeout, processor.getRequestProcessor().getClient().callTimeoutMillis()); + assertEquals(3_000, processor.getClient().callTimeoutMillis()); + assertFalse(processor.getClient().retryOnConnectionFailure()); + // Three attempts timing out, after waiting up to 1s, then 2s, before the retries. + assertEquals(3 * 3_000 + 1_000 + 2_000, processor.getCloseTimeoutMillis()); processor.close(); } + @Test @SneakyThrows - public void close_sendsTheLastBatchAlongsideTheBatchesInFlight() { + public void close_sendsTheEventsBufferedBehindTheBatchInFlight() { AcceptingInterceptor eventsApi = AcceptingInterceptor.blocked(); EventProcessor processor = newProcessor(1, 0, eventsApi); eventProcessor = null; FlagsmithLogger logger = mockLogger(processor); - for (int i = 0; i <= EventProcessor.MAX_IN_FLIGHT_BATCHES; i++) { - processor.trackEvent("purchase", "user-" + i, "1", null, null); - } - assertTrue(recorder.awaitCount(EventProcessor.MAX_IN_FLIGHT_BATCHES)); + processor.trackEvent("purchase", "user-0", "1", null, null); + processor.trackEvent("purchase", "user-1", "1", null, null); + assertTrue(recorder.awaitCount(1)); assertEquals(1, processor.bufferedEvents().size()); CompletableFuture closed = CompletableFuture.runAsync(processor::close); - assertTrue(recorder.awaitCount(EventProcessor.MAX_IN_FLIGHT_BATCHES + 1)); eventsApi.release(); closed.get(WAIT_SECONDS, TimeUnit.SECONDS); - assertEquals(EventProcessor.MAX_IN_FLIGHT_BATCHES + 1, deliveredEvents()); + assertEquals(2, recorder.count()); + assertEquals(2, deliveredEvents()); assertEquals(Collections.emptyList(), errorCalls(logger)); } @@ -602,32 +720,63 @@ public void close_stopsWaitingAtTheTimeout() { @Test @SneakyThrows - public void settle_sendsEventsThatWaitedBehindTheInFlightLimit() { + public void close_abandonsTheRequestInFlightAtItsTimeout() { + CountDownLatch requested = new CountDownLatch(1); + CompletableFuture abandoned = new CompletableFuture<>(); + EventProcessor processor = newProcessor(1000, 0, (chain) -> { + requested.countDown(); + try { + Thread.sleep(TimeUnit.SECONDS.toMillis(WAIT_SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + abandoned.complete(null); + throw new IOException("Canceled"); + }); + eventProcessor = null; + mockLogger(processor); + + processor.trackEvent("purchase", "user-1", "1", null, null); + processor.flush(); + assertTrue(requested.await(WAIT_SECONDS, TimeUnit.SECONDS)); + long start = System.nanoTime(); + processor.close(300); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + + assertTrue(elapsedMillis < 2_000, "close(300) waited " + elapsedMillis + "ms"); + abandoned.get(WAIT_SECONDS / 2, TimeUnit.SECONDS); + assertTrue(processor.getSender().awaitTermination(WAIT_SECONDS, TimeUnit.SECONDS)); + assertEquals(1, recorder.count()); + assertTrue(processor.bufferedEvents().isEmpty()); + } + + @Test + @SneakyThrows + public void settle_sendsEventsThatWaitedBehindTheBatchInFlight() { AcceptingInterceptor eventsApi = AcceptingInterceptor.blocked(); EventProcessor processor = newProcessor(1000, 0, eventsApi); FlagsmithLogger logger = mockLogger(processor); - for (int i = 0; i < EventProcessor.MAX_IN_FLIGHT_BATCHES; i++) { - processor.trackEvent("purchase", "user-" + i, "1", null, null); - processor.flush(); - } + processor.trackEvent("purchase", "user-0", "1", null, null); + processor.flush(); for (int i = 0; i < 100; i++) { processor.trackEvent("purchase", "waiting-" + i, "1", null, null); processor.flush(); } - assertTrue(recorder.awaitCount(EventProcessor.MAX_IN_FLIGHT_BATCHES)); + assertTrue(recorder.awaitCount(1)); assertEquals(100, processor.bufferedEvents().size()); eventsApi.release(); - assertTrue(awaitDelivered(EventProcessor.MAX_IN_FLIGHT_BATCHES + 100)); + assertTrue(awaitDelivered(1 + 100)); + assertEquals(2, recorder.count()); assertEquals(Collections.emptyList(), errorCalls(logger)); } @Test @SneakyThrows - public void flush_waitsForTheEventsBufferedBehindTheInFlightLimit() { + public void flush_waitsForTheEventsBufferedBehindTheBatchInFlight() { CountDownLatch inFlightReleased = new CountDownLatch(1); CountDownLatch waitingReleased = new CountDownLatch(1); EventProcessor processor = newProcessor(1000, 0, (chain) -> { @@ -643,30 +792,28 @@ public void flush_waitsForTheEventsBufferedBehindTheInFlightLimit() { return AcceptingInterceptor.accepted(chain.request()); }); - for (int i = 0; i < EventProcessor.MAX_IN_FLIGHT_BATCHES; i++) { - processor.trackEvent("purchase", "user-" + i, "1", null, null); - processor.flush(); - } + processor.trackEvent("purchase", "user-0", "1", null, null); + processor.flush(); processor.trackEvent("purchase", "waiting", "1", null, null); CompletableFuture flushed = processor.flush(); - assertTrue(recorder.awaitCount(EventProcessor.MAX_IN_FLIGHT_BATCHES)); + assertTrue(recorder.awaitCount(1)); inFlightReleased.countDown(); - assertTrue(recorder.awaitCount(EventProcessor.MAX_IN_FLIGHT_BATCHES + 1)); + assertTrue(recorder.awaitCount(2)); assertFalse(flushed.isDone()); waitingReleased.countDown(); flushed.get(WAIT_SECONDS, TimeUnit.SECONDS); - assertEquals(EventProcessor.MAX_IN_FLIGHT_BATCHES + 1, deliveredEvents()); + assertEquals(2, deliveredEvents()); } @Test @SneakyThrows - public void trackEvent_dropsTheOldestEventsOnceTheBufferIsFullBehindTheLimit() { + public void trackEvent_dropsTheOldestEventsOnceTheBufferIsFull() { AcceptingInterceptor eventsApi = AcceptingInterceptor.blocked(); EventProcessor processor = newProcessor(1000, 0, eventsApi); FlagsmithLogger logger = mockLogger(processor); - int inFlight = EventProcessor.MAX_IN_FLIGHT_BATCHES * 1000; + int inFlight = 1000; for (int i = 0; i < inFlight; i++) { processor.trackEvent("purchase", "user-" + i, "1", null, null); @@ -690,6 +837,30 @@ public void trackEvent_dropsTheOldestEventsOnceTheBufferIsFullBehindTheLimit() { assertTrue(bodies.contains("overflow-500-")); } + @Test + @SneakyThrows + public void trackEvent_dropsTheOldestKeptEventOnceTheBufferIsFull() { + EventProcessor processor = newProcessor(1000, 0, failingWith(503, 503, 503)); + mockLogger(processor); + interceptor.addRule().post(EVENTS_ENDPOINT).anyTimes().respond(ACCEPTED_BODY, MEDIATYPE_JSON); + + for (int i = 1; i <= EventProcessor.MAX_BUFFERED_EVENTS; i++) { + processor.trackEvent("event_" + i, "user-1", null, null, null); + } + flushAndWait(processor); + assertEquals(EventProcessor.MAX_BUFFERED_EVENTS, processor.bufferedEvents().size()); + + processor.trackEvent("event_1001", "user-1", null, null, null); + flushAndWait(processor); + + assertEquals(EventProcessor.MAX_ATTEMPTS + 1, recorder.count()); + JsonNode events = MapperFactory.getMapper() + .readTree(recorder.bodies().get(EventProcessor.MAX_ATTEMPTS)).get("events"); + assertEquals(EventProcessor.MAX_BUFFERED_EVENTS, events.size()); + assertEquals("event_2", events.get(0).get("event").asText()); + assertEquals("event_1001", events.get(events.size() - 1).get("event").asText()); + } + @Test @SneakyThrows public void trackExposureEvent_buffersAnExposureAgainOnceItsCopyWasDropped() { @@ -697,7 +868,7 @@ public void trackExposureEvent_buffersAnExposureAgainOnceItsCopyWasDropped() { EventProcessor processor = newProcessor(1000, 0, eventsApi); mockLogger(processor); - for (int i = 0; i < EventProcessor.MAX_IN_FLIGHT_BATCHES * 1000; i++) { + for (int i = 0; i < 1000; i++) { processor.trackEvent("purchase", "user-" + i, "1", null, null); } processor.trackExposureEvent("checkout_cta", "user-1", "treatment", null, null);