Merge pull request #226 from splitio/size-event-queue

Size event queue

Author: ldecheverz-split

client/src/main/java/io/split/client/EventClientImpl.java

@@ -54,23 +54,15 @@ public class EventClientImpl implements EventClient {
     private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
 
     ThreadFactory eventClientThreadFactory(final String name) {
-        return new ThreadFactory() {
-            @Override
-            public Thread newThread(final Runnable r) {
-                return new Thread(new Runnable() {
-                    @Override
-                    public void run() {
-                        Thread.currentThread().setPriority(MIN_PRIORITY);
-                        r.run();
-                    }
-                }, name);
-            }
-        };
+        return r -> new Thread(() -> {
+            Thread.currentThread().setPriority(MIN_PRIORITY);
+            r.run();
+        }, name);
     }
 
 
     public static EventClientImpl create(CloseableHttpClient httpclient, URI eventsRootTarget, int maxQueueSize, long flushIntervalMillis, int waitBeforeShutdown, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
-        return new EventClientImpl(new LinkedBlockingQueue<WrappedEvent>(),
+        return new EventClientImpl(new LinkedBlockingQueue<>(maxQueueSize),
                 httpclient,
                 Utils.appendPath(eventsRootTarget, "api/events/bulk"),
                 maxQueueSize,
@@ -131,10 +123,14 @@ public boolean track(Event event, int eventSize) {
             if (event == null) {
                 return false;
             }
-            _eventQueue.put(new WrappedEvent(event, eventSize));
-            _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_QUEUED, 1);
+            if(_eventQueue.offer(new WrappedEvent(event, eventSize))) {
+                _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_QUEUED, 1);
+            }
+            else {
+                _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_DROPPED, 1);
+            }
 
-        } catch (ClassCastException | NullPointerException | InterruptedException e) {
+        } catch (ClassCastException | NullPointerException | IllegalArgumentException e) {
             _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_DROPPED, 1);
             _log.warn("Interruption when adding event withed while adding message %s.", event);
             return false;
@@ -164,7 +160,7 @@ public void run() {
             List<Event> events = new ArrayList<>();
             long accumulated = 0;
             try {
-                while (true) {
+                while (!Thread.currentThread().isInterrupted()) {
                     WrappedEvent data = _eventQueue.take();
                     Event event = data.event();
                     Long size = data.size();

client/src/test/java/io/split/client/EventsClientImplTest.java

@@ -1,13 +1,15 @@
 package io.split.client;
 
 import io.split.client.dtos.Event;
+import io.split.telemetry.domain.enums.EventsDataRecordsEnum;
 import io.split.telemetry.storage.InMemoryTelemetryStorage;
 import io.split.telemetry.storage.TelemetryStorage;
 import org.apache.hc.client5.http.classic.methods.HttpUriRequest;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.HttpClients;
 import org.hamcrest.Matchers;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
 
@@ -74,4 +76,25 @@ public void testEventsFlushedWhenSizeLimitReached() throws URISyntaxException, I
         Thread.sleep(2000);
         Mockito.verify(client, Mockito.times(1)).execute((HttpUriRequest) Mockito.any());
     }
+
+    @Test
+    public void testEventDropped() throws URISyntaxException, NoSuchFieldException, IllegalAccessException, InterruptedException {
+        TelemetryStorage telemetryStorage = Mockito.mock(InMemoryTelemetryStorage.class);
+        CloseableHttpClient client = Mockito.mock(CloseableHttpClient.class);
+        EventClientImpl eventClient = new EventClientImpl(new LinkedBlockingQueue<>(2),
+                client,
+                URI.create("https://kubernetesturl.com/split"),
+                10000, // Long queue so it doesn't flush by # of events
+                100000, // Long period so it doesn't flush by timeout expiration.
+                0, telemetryStorage);
+        eventClient.close();
+        Thread.sleep(1000);
+        for (int i = 0; i < 3; ++i) {
+            Event event = new Event();
+            eventClient.track(event, 1);
+        }
+
+        Mockito.verify(telemetryStorage, Mockito.times(2)).recordEventStats(EventsDataRecordsEnum.EVENTS_QUEUED, 1);
+        Mockito.verify(telemetryStorage, Mockito.times(1)).recordEventStats(EventsDataRecordsEnum.EVENTS_DROPPED, 1);
+    }
 }

client/src/test/java/io/split/engine/segments/SegmentSynchronizationTaskImpTest.java

@@ -106,8 +106,7 @@ public void testFetchAllAsynchronousAndGetFalse() throws NoSuchFieldException, I
         modifiersField.setAccessible(true);
         modifiersField.setInt(segmentFetchersForced, segmentFetchersForced.getModifiers() & ~Modifier.FINAL);
 
-        segmentFetchersForced.set(fetchers, _segmentFetchers); // 1ms
-        fetcher1.set(segmentFetcher);
+        segmentFetchersForced.set(fetchers, _segmentFetchers);
         boolean fetch = fetchers.fetchAllSynchronous();
         Assert.assertEquals(false, fetch);
     }
@@ -117,6 +116,7 @@ public void testFetchAllAsynchronousAndGetTrue() throws NoSuchFieldException, Il
         SDKReadinessGates gates = new SDKReadinessGates();
         SegmentCache segmentCache = Mockito.mock(SegmentCache.class);
 
+        ConcurrentMap<String, SegmentFetcher> _segmentFetchers = Maps.newConcurrentMap();
         SegmentChangeFetcher segmentChangeFetcher = Mockito.mock(SegmentChangeFetcher.class);
         SegmentFetcherImp segmentFetcher = Mockito.mock(SegmentFetcherImp.class);
         final SegmentSynchronizationTaskImp fetchers = new SegmentSynchronizationTaskImp(segmentChangeFetcher, 1L, 1, gates, segmentCache, TELEMETRY_STORAGE);
@@ -127,6 +127,7 @@ public void testFetchAllAsynchronousAndGetTrue() throws NoSuchFieldException, Il
         Field modifiersField = Field.class.getDeclaredField("modifiers");
         modifiersField.setAccessible(true);
         modifiersField.setInt(segmentFetchersForced, segmentFetchersForced.getModifiers() & ~Modifier.FINAL);
+        segmentFetchersForced.set(fetchers, _segmentFetchers);
         Mockito.doNothing().when(segmentFetcher).callLoopRun(Mockito.anyBoolean(),Mockito.anyBoolean());
         Mockito.when(segmentFetcher.runWhitCacheHeader()).thenReturn(true);
         Mockito.when(segmentFetcher.fetchAndUpdate(Mockito.anyBoolean())).thenReturn(true);