Configurable max queue size

[wadejensen]

It would be really great if analytics-java supported the ability to configure the maximum number of inflight messages. The current queues within analytics-java are unbounded. This would help prevent running out of memory due to a high volume of messages if the Segment API or our proxy to it became unavailable.

At our company we send large volumes of messages to AWS Kinesis Firehose from our backends. I would like to route them through Segment instead, however I'm not comfortable firing that volume of events through this client on all our backends without a method of circuit-breaking or dropping messages in the case of a downstream outage.

It's already technically possible to configure the client's networkExecutor to drop batches of messages beyond some threshold (example to follow in a comment). However that configuration code is well, a lot... And it seems less than ideal to drop entire batches in that scenario.

I took a run at writing something myself, but quickly realised that because of the three threads in play: the looperExecutor, networkExecutor, and flushScheduler, its not as simple as placing a limit on the capacity of the initial messageQueue, since the looper is busy-waiting to pull messages off, batch them, and submit to the networkExecutor.

I see a few ways to support this feature, however when I played around with this, none of them were a surgical, light touch PR, and testing wasn't smooth, so I wanted to get a sense of the interest in accepting a PR for this feature and what approach would be preferred.

Possible approaches:

• Smallest change. Allowing a limit on the depth of messageQueue and use a SynchronousQueue for the networkExecutor. However this would effectively block the looperExecutor on submission to the networkExecutor and arguably defeat the purpose of a separate thread, and allowing the configuration of a networkExecutor.
• Allowing a limit on the depth of messageQueue and also using BlockingQueue with a very small limit for the networkExecutor.
• Allowing a limit on the depth of messageQueue and also calling SegmentService:upload directly from looperExecutor and blocking the busy-waiting loop, ie. merge the looperExecutor and networkExecutor.

Everything I'm proposing here seems to be fighting against a separate looper and networkExecutor thread. I'm keen to understand the motivation for the current design so I could operate better within the spirit of it. Given the default networkExecutor instance is single-threaded, I'm curious how many people use a larger thread pool which would take advantage of this decoupling.

Apologies for opening what is essentially a duplicate to #147. However that issue has been inactive for a long time and proposes a PR that doesn't entirely resolve the issue.

Please let me know what changes you'd be open to 🙂

[wadejensen]

Example configuration which should work with the existing functionality to limit memory usage by dropping batches with the RejectedExecutionHandler DiscardPolicy.

    Preconditions.checkArgument(batchSize < queueSize, "batchSize must be less than queueSize.");
    int batchQueueSize = queueSize / batchSize;

    var client = Analytics
      .builder(segmentWriteKey)
      .flushQueueSize(batchSize)
      .flushInterval(flushIntervalSeconds, TimeUnit.SECONDS)
      .networkExecutor(networkExecutor(batchQueueSize))
      .build();

    return new SegmentAnalyticsService(client);
  }

  /**
   * Executor for submitting message batches to an endpoint.
   * Will drop excess message batches and log an error when message queue size exceeds
   * {@code batchQueueSize}.
   *
   * @param maxQueueSize maximum number of message batches to be queued in memory
   */
  private static ExecutorService networkExecutor(int batchQueueSize) {
    var workQueue = new LinkedBlockingQueue<Runnable>(batchQueueSize);
    var threadFactory = lowPriorityThreadFactory();
    return new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, workQueue,
      threadFactory, new DiscardPolicy());
  }

  private static ThreadFactory lowPriorityThreadFactory() {
    return r -> new Thread(() -> {
      Thread.currentThread().setPriority(MIN_PRIORITY);
      r.run();
    }, THREAD_NAME);
  }

[wadejensen]

I've taken a stab at the first approach listed above. PTAL
#162

[wadejensen]

We managed to put in a hacky solution to resolve the memory leak.

var client = Analytics.builder(segmentWriteKey)
        .networkExecutor(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(256),
            (r, executor) -> logger.warn("discarding analytics batch")))
        .build()

its not great because it drops batches of events rather than individual ones when the network queue is full, but it doesn't require source changes and is good enough for our current purposes.