This function is only called when sender is closed forcefully. It will fail all the
        incomplete batches and return.
778615
def abort_incomplete_batches(self):
        
        # We need to keep aborting the incomplete batch until no thread is trying to append to
        # 1. Avoid losing batches.
        # 2. Free up memory in case appending threads are blocked on buffer full.
        # This is a tight loop but should be able to get through very quickly.
        while True:
            self._abort_batches()
            if not self._appends_in_progress.get():
                break
        # After this point, no thread will append any messages because they will see the close
        # flag set. We need to do the last abort after no thread was appending in case the there was a new
        # batch appended by the last appending thread.
        self._abort_batches()
        self._batches.clear()
728418
async def close(self, *, force_after=30):
        ''''''
        if self.transport:
            self.transport.close()
            try:
                async with timeout_after(force_after):
                    await self.closed_event.wait()
            except TaskTimeout:
                self.abort()
                await self.closed_event.wait()
878583
@Override
  public void close() throws IOException {
    try {
      if (consumer != null && consumer instanceof Closeable) {
        // Call close in a new transaction.
        // TODO (terence): Actually need to coordinates with other flowlets to drain the queue.
        TransactionContext txContext = dataFabricFacade.createTransactionManager();
        txContext.start();
        try {
          ((Closeable) consumer).close();
          txContext.finish();
        } catch (TransactionFailureException e) {
          LOG.warn("Fail to commit transaction when closing consumer.");
          txContext.abort();
        }
      }
    } catch (Exception e) {
      LOG.warn("Fail to close queue consumer.", e);
    }
    consumer = null;
  }
1123533
public void onFailure(Throwable t)
    {
        MessagingService.instance().sendOneWay(new SyncComplete(desc, request.src, request.dst, false).createMessage(), request.initiator);
    }
156295
function _endBatch() {
        --_batch;
        _cometd._debug('Ending batch, depth', _batch);
        if (_batch < 0) {
            throw 'Calls to startBatch() and endBatch() are not paired';
        }

        if (_batch === 0 && !_isDisconnected() && !_internalBatch) {
            _flushBatch();
        }
    }
1179967
@Override
  public FailsafeExecutor<R> onFailure(CheckedConsumer<? extends ExecutionCompletedEvent<R>> listener) {
    return super.onFailure(listener);
  }
1293861
private void processFailedReceivedDeleted(long[] failedMap, Block[] sent) {
    synchronized (receivedAndDeletedBlockList) {
      // Blocks that do not belong to an Inode are saved for
      // retransmisions
      for (int i = sent.length - 1 ; i >= 0; i--) {
        if(!LightWeightBitSet.get(failedMap, i)){
          continue;
        }
        // Insert into retry list.
        LOG.info("Block " + sent[i] + " does not belong to any file "
            + "on namenode " + avatarnodeAddress + " Retry later.");
        receivedAndDeletedBlockList.add(0, sent[i]);
        if (!DFSUtil.isDeleted(sent[i])) {
          pendingReceivedRequests++;
        }
      }
      lastBlockReceivedFailed = AvatarDataNode.now();
    }
  }
846175
def _on_closed(self):
        """"""
        LOGGER.error('Redis connection closed')
        self.connected = False
        self._on_close()
        self._stream = None
1249375
def maybe_call_closing_deferred(self):
        

        if self._closing_deferred:
            self._closing_deferred.callback(self)
            self._closing_deferred = None
        self._when_closed.fire(self)
358622
public void afterCompleted(Event event) {
		ChaincodeMessage message = messageHelper(event);
		logger.debug(String.format("[%s]sending COMPLETED to validator for tid", shortID(message)));
		try {
			serialSend(message);
		} catch (Exception e) {
			event.cancel(new Exception("send COMPLETED failed %s", e));
		}
	}
354144
private synchronized void complete(String message) {
    
    // This is a weird problem with writing stuff while idling. Need to investigate it more, but
    // for now just ignore it.
    if (MESSAGE_COULDNT_BE_FETCHED_REGEX.matcher(message).matches()) {
      log.warn("Some messages in the batch could not be fetched for {}\n" +
          "---cmd---\n{}\n---wire---\n{}\n---end---\n", new Object[] {
          config.getUsername(),
          getCommandTrace(),
          getWireTrace()
      });
      errorStack.push(new Error(completions.peek(), message, wireTrace.list()));
      final CommandCompletion completion = completions.peek();
      String errorMsg = "Some messages in the batch could not be fetched for user " + config.getUsername();
      RuntimeException ex = new RuntimeException(errorMsg);
      if (completion != null) {
        completion.error(errorMsg, new MailHandlingException(getWireTrace(), errorMsg, ex));
        completions.poll();
      } else {
        throw ex;
      }
    }

    CommandCompletion completion = completions.peek();
    if (completion == null) {
      if ("+ idling".equalsIgnoreCase(message)) {
        synchronized (idleMutex) {
          idler.idleStart();
          log.trace("IDLE entered.");
          idleAcknowledged.set(true);
        }
      } else {
        log.error("Could not find the completion for message {} (Was it ever issued?)", message);
        errorStack.push(new Error(null, "No completion found!", wireTrace.list()));
      }
      return;
    }
    
    if (completion.complete(message)) {
      completions.poll();
    }
  }
1194835
@Override
    public void operationComplete(ChannelFuture future) throws Exception {
        // If a message has been sent out, cancel the timeout for starting a request.
        cancelTimeout();

        if (future.isSuccess()) {
            // The first write is always the first headers, so log that we finished our first transfer over the
            // wire.
            if (!loggedRequestFirstBytesTransferred) {
                logBuilder.requestFirstBytesTransferred();
                loggedRequestFirstBytesTransferred = true;
            }

            if (state == State.DONE) {
                // Successfully sent the request; schedule the response timeout.
                response.scheduleTimeout(ch.eventLoop());
            }

            // Request more messages regardless whether the state is DONE. It makes the producer have
            // a chance to produce the last call such as 'onComplete' and 'onError' when there are
            // no more messages it can produce.
            if (!isSubscriptionCompleted) {
                assert subscription != null;
                subscription.request(1);
            }
            return;
        }

        fail(future.cause());

        final Throwable cause = future.cause();
        if (!(cause instanceof ClosedPublisherException)) {
            final Channel ch = future.channel();
            Exceptions.logIfUnexpected(logger, ch, HttpSession.get(ch).protocol(), cause);
            ch.close();
        }
    }
65450
def on_consumer_cancelled(self, method_frame):
        
        msg = 'Consumer was cancelled remotely, shutting down: {0!r}'
        logger.info(msg.format(method_frame))
        if self._channel:
            self._channel.close()
1044329
private S waitForCompletionAndThrowIfUploadFailed() throws IOException {
    try {
      return uploadOperation.get();
    } catch (InterruptedException e) {
      // If we were interrupted, we need to cancel the upload operation.
      uploadOperation.cancel(true);
      IOException exception = new ClosedByInterruptException();
      exception.addSuppressed(e);
      throw exception;
    } catch (ExecutionException e) {
      if (e.getCause() instanceof Error) {
        throw (Error) e.getCause();
      }
      throw new IOException("Upload failed", e.getCause());
    }
  }
661434
public void processFailure(QueryBatchException queryBatch) {
    boolean isHostUnavailableException = processException(queryBatch.getBatcher(), queryBatch, queryBatch.getClient().getHost());
    if ( isHostUnavailableException == true ) {
      try {
        logger.warn("Retrying failed batch: {}, results so far: {}, forest: {}, forestBatch: {}, forest results so far: {}",
          queryBatch.getJobBatchNumber(), queryBatch.getJobResultsSoFar(), queryBatch.getForest().getForestName(),
          queryBatch.getForestBatchNumber(), queryBatch.getForestResultsSoFar());
        queryBatch.getBatcher().retryWithFailureListeners(queryBatch);
      } catch (RuntimeException e) {
        logger.error("Exception during retry", e);
        processFailure(new QueryBatchException(queryBatch, e));
      }
    }
  }
293331
@Override
	public void close() throws IOException {
		try {
			upload.executeBatch();
			batchCount = 0;
		} catch (SQLException se) {
			throw new IllegalArgumentException("close() failed", se);
		} catch (NullPointerException se) {
		}
		try {
			upload.close();
		} catch (SQLException se) {
			LOG.info("Inputformat couldn't be closed - " + se.getMessage());
		} catch (NullPointerException npe) {
		}
		try {
			dbConn.close();
		} catch (SQLException se) {
			LOG.info("Inputformat couldn't be closed - " + se.getMessage());
		} catch (NullPointerException npe) {
		}
	}
1355067
function onResultStreamClose(err, continueCallback)
  {
    // if the error is retryable and
    // the result stream hasn't been closed too many times
    if (isResultStreamErrorRetryable(err) &&
        (numResultStreamInterrupts <
        context.connectionConfig.getResultStreamInterrupts()))
    {
      numResultStreamInterrupts++;

      // fetch the statement result again
      context.refresh(function()
      {
        if (context.resultError)
        {
          close(context.resultError);
        }
        else
        {
          continueCallback();
        }
      });
    }
    else
    {
      close(err);
    }
  }