Class BatchInserter

storm-client/src/jvm/org/apache/storm/utils/JCQueue.java:326–410

    /* Not thread safe. Have one instance per producer thread or synchronize externally */
    private static class BatchInserter implements Inserter {
        private final int batchSz;
        private JCQueue queue;
        private ArrayList<Object> currentBatch;

        BatchInserter(JCQueue queue, int batchSz) {
            this.queue = queue;
            this.batchSz = batchSz;
            this.currentBatch = new ArrayList<>(batchSz + 1);
        }

        /**
         * Blocking call - retires till element is successfully added.
         */
        @Override
        public void publish(Object obj) throws InterruptedException {
            currentBatch.add(obj);
            if (currentBatch.size() >= batchSz) {
                flush();
            }
        }

        /**
         * Non-Blocking call. return value indicates success/failure
         */
        @Override
        public boolean tryPublish(Object obj) {
            if (currentBatch.size() >= batchSz) {
                if (!tryFlush()) {
                    return false;
                }
            }
            currentBatch.add(obj);
            return true;
        }

        /**
         * Blocking call - Does not return until at least 1 element is drained or Thread.interrupt() is received. Uses backpressure wait
         * strategy.
         */
        @Override
        public void flush() throws InterruptedException {
            if (currentBatch.isEmpty()) {
                return;
            }
            int publishCount = queue.tryPublishInternal(currentBatch);
            int retryCount = 0;
            while (publishCount == 0) { // retry till at least 1 element is drained
                for (JCQueueMetrics jcQueueMetric : queue.jcqMetrics) {
                    jcQueueMetric.notifyInsertFailure();
                }
                if (retryCount == 0) { // check avoids multiple log msgs when in a idle loop
                    LOG.debug("Experiencing Back Pressure when flushing batch to Q: '{}'. Entering BackPressure Wait.",
                        queue.getQueueName());
                }
                retryCount = queue.backPressureWaitStrategy.idle(retryCount);
                if (Thread.interrupted()) {
                    throw new InterruptedException();