code: void onSealExclusive(Batch batch, long elapsedTimeMillis)
    {
      batchFillingTimeCounter.add((int) Math.max(elapsedTimeMillis, 0));
      if (elapsedTimeMillis > 0) {
        // If elapsedTimeMillis is 0 or negative, it's likely because System.currentTimeMillis() is not monotonic, so not
        // accounting this time for determining batch sending timeout.
        lastFillTimeMillis = elapsedTimeMillis;
      }
      addBatchToEmitQueue(batch);
      wakeUpEmittingThread();
      if (!isTerminated()) {
        int nextBatchNumber = EmittedBatchCounter.nextBatchNumber(batch.batchNumber);
        if (!concurrentBatch.compareAndSet(batch, new Batch(this, acquireBuffer(), nextBatchNumber))) {
          // If compareAndSet failed, the service is closed concurrently.
          Preconditions.checkState(isTerminated());
        }
      }
    }
docs: Called from {@link Batch} only once for each Batch in existence.