When close is invoked, all new coming records will be rejected
Add a busy loop here to ensure all the ongoing appends are completed
320581
public void close () {
    closed = true;
    while (appendsInProgress.get() > 0) {
      LOG.info("Append is still going on, wait for a while");
      try {
        Thread.sleep(100);
      } catch (InterruptedException e) {
        LOG.error("close is interrupted while appending data is in progress");
      }
    }
    this.closeComplete.countDown();
  }
469199
function () {
        var that = this,
            fullQuery = that.extendedCurrentValue(),
            currentValue = that.el.val(),
            resolver = $.Deferred();

        resolver
            .done(function (suggestion) {
                that.selectSuggestion(suggestion, 0, currentValue, { hasBeenEnriched: true });
                that.el.trigger('suggestions-fixdata', suggestion);
            })
            .fail(function () {
                that.selection = null;
                that.el.trigger('suggestions-fixdata');
            });

        if (that.isQueryRequestable(fullQuery)) {
            that.currentValue = fullQuery;
            that.getSuggestions(fullQuery, { count: 1, from_bound: null, to_bound: null })
                .done(function (suggestions) {
                    // data fetched
                    var suggestion = suggestions[0];
                    if (suggestion) {
                        resolver.resolve(suggestion);
                    } else {
                        resolver.reject();
                    }
                })
                .fail(function () {
                    // no data fetched
                    resolver.reject();
                });
        } else {
            resolver.reject();
        }
    }
187471
@Override
    public void executeEventCallback(Runnable callback) {
        if (!nodeEngine.isRunning()) {
            return;
        }
        try {
            eventExecutor.execute(callback);
        } catch (RejectedExecutionException e) {
            rejectedCount.inc();

            if (eventExecutor.isLive()) {
                logFailure("EventQueue overloaded! Failed to execute event callback: %s", callback);
            }
        }
    }
515258
public static UpdateOnCloseHandler addUpdateOnCloseHandler(Record record, Record recordToUpdate, boolean bRefreshAfterUpdate, boolean bUpdateOnClose, boolean bUpdateOnUpdate)
    {
        UpdateOnCloseHandler listener = (UpdateOnCloseHandler)record.getListener(UpdateOnCloseHandler.class);
        while (listener != null)
        {
            if (listener.getRecordToUpdate() == recordToUpdate)
                break;  // Found, don't add again
            listener = (UpdateOnCloseHandler)listener.getListener(UpdateOnCloseHandler.class);
        }
        if (listener == null)
            record.addListener(listener = new UpdateOnCloseHandler(recordToUpdate, bRefreshAfterUpdate, bUpdateOnClose, bUpdateOnUpdate)); // Make sure this is updated with booking
        return listener;
    }
444014
@Override
  public void onCompleted() {
    RpcUtils.streamingRPCAndLog(LOG, new RpcUtils.StreamingRpcCallable<OpenLocalBlockResponse>() {
      @Override
      public OpenLocalBlockResponse call() throws Exception {
        if (mLockId != BlockLockManager.INVALID_LOCK_ID) {
          mWorker.unlockBlock(mLockId);
          mLockId = BlockLockManager.INVALID_LOCK_ID;
        } else if (mRequest != null) {
          LOG.warn("Close a closed block {}.", mRequest.getBlockId());
        }
        return null;
      }

      @Override
      public void exceptionCaught(Throwable e) {
        mResponseObserver.onError(GrpcExceptionUtils.fromThrowable(e));
        mLockId = BlockLockManager.INVALID_LOCK_ID;
      }
    }, "CloseBlock", false, true, mResponseObserver, "Session=%d, Request=%s",
        mSessionId, mRequest);
  }
1022548
def wait_closed(self):
        
        if self._closed:
            return
        if not self._closing:
            raise RuntimeError(
                ".wait_closed() should be called "
                "after .close()"
            )

        while self._free:
            conn = self._free.popleft()
            if not conn.closed:
                yield from conn.close()
            else:
                # pragma: no cover
                pass
        with (yield from self._cond):
            while self.size > self.freesize:
                yield from self._cond.wait()
        self._used.clear()
        self._closed = True
1219890
def waitForAllConnectionsToClose(self):
        
        if not self._connections:
            return self._stop()
        return self._allConnectionsClosed.deferred().addBoth(self._stop)
969175
private void doClose() {
    open = false;
    connections.remove(this);

    for (Map.Entry<Long, ContextualFuture> entry : futures.entrySet()) {
      ContextualFuture future = entry.getValue();
      try {
        future.context.executor().execute(() -> future.completeExceptionally(new ConnectException("connection closed")));
      } catch (RejectedExecutionException e) {
      }
    }
    futures.clear();

    for (Consumer<Connection> closeListener : closeListeners) {
      try {
        context.executor().execute(() -> closeListener.accept(this));
      } catch (RejectedExecutionException e) {
      }
    }
  }
891590
void endOffering(IOException exception) {
    if (closed) {
      return;
    }

    // closed should never be set by this method--only the polling side should do it, as it means
    // that the CLOSED marker has been encountered.

    if (this.exception == null) {
      this.exception = exception;
    }

    // Since this is an unbounded queue, offer() always succeeds.
    queue.offer(CLOSED);
  }
832825
@Override
	public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
		int counts = rejectCounts.incrementAndGet();
		if(logger.isWarnEnabled()) {
			int t = counts % 100;
			if (t == 0)
			{
				logger.warn(new StringBuilder().append("Task[Import DB Data to Elasticsearch] rejected  ").append(counts).append(" times.").toString());
			}
		}
//
		try {
			executor.getQueue().put(r);
		} catch (InterruptedException e1) {
			throw new RejectedExecutionException(e1);
		}
	}
720698
def push(self, value):
        
        if self.closed and not self.allow_add_after_close:
            Log.error("Do not push to closed queue")

        with self.lock:
            self._wait_for_queue_space()
            if not self.closed:
                self.queue.appendleft(value)
        return self
6973
def _AppendRecord(self):
    """"""

    # If no Values then don't output.
    if not self.values:
      return

    cur_record = []
    for value in self.values:
      try:
        value.OnSaveRecord()
      except SkipRecord:
        self._ClearRecord()
        return
      except SkipValue:
        continue

      # Build current record into a list.
      cur_record.append(value.value)

    # If no Values in template or whole record is empty then don't output.
    if len(cur_record) == (cur_record.count(None) + cur_record.count([])):
      return

    # Replace any 'None' entries with null string ''.
    while None in cur_record:
      cur_record[cur_record.index(None)] = ''

    self._result.append(cur_record)
    self._ClearRecord()
533379
public void waitUntilComplete() {
        lock.lock();
        try {
            while(processing || count.get() > 0) {
                try {
                    processing_done.await();
                }
                catch(InterruptedException ignored) {
                }
            }
        }
        finally {
            lock.unlock();
        }
    }
381999
function updateQueryForNextPage(pulledRecords) {
        return Platform.async(function(callback) {
            callback();
        })().then(function() {

            if (!pulledRecords) {
                throw new Error('Something is wrong. pulledRecords cannot be null at this point');
            }

            var lastRecord = pulledRecords[ pulledRecords.length - 1];

            if (!lastRecord) {
                throw new Error('Something is wrong. Possibly invalid response from the server. lastRecord cannot be null!');
            }

            var lastRecordTime = lastRecord[tableConstants.sysProps.updatedAtColumnName];

            if (!_.isDate(lastRecordTime)) {
                throw new Error('Property ' + tableConstants.sysProps.updatedAtColumnName + ' of the last record should be a valid date');
            }

            if (lastRecordTime.getTime() === lastKnownUpdatedAt.getTime()) {
                pagePullQuery.skip(pagePullQuery.getComponents().skip + pulledRecords.length);
            } else {
                buildQueryFromLastKnownUpdateAt(lastRecordTime);
            }
        });
    }
1090748
def shutdown
      fail EventQueueDeadError, 'event queue already dead' if @dead

      queue = []
      @mutex.synchronize do
        @dead = true
        @not_empty.broadcast

        queue = @queue
        @queue.clear
      end
      while !queue.empty?
        discard_message(queue.pop)
      end
      true
    end
518340
function(curr) {
            if (curr.state === STATE_FULFILLED) execute_handlers(curr, "onFulfilled", curr.fulfillValue); else if (curr.state === STATE_REJECTED) execute_handlers(curr, "onRejected", curr.rejectReason);
        }
334922
@Override
    public void append(Append append) {
        log.trace("Processing append received from client {}", append);
        UUID id = append.getWriterId();
        synchronized (lock) {
            Long lastEventNumber = latestEventNumbers.get(Pair.of(append.getSegment(), id));
            Preconditions.checkState(lastEventNumber != null, "Data from unexpected connection: %s.", id);
            Preconditions.checkState(append.getEventNumber() >= lastEventNumber, "Event was already appended.");
            waitingAppends.put(id, append);
        }
        pauseOrResumeReading();
        performNextWrite();
    }