int totalBytesRead = 0;
    boolean doneEncoding = endOfInput;

    DRAINING:
    while (true) {
      // We stay in draining mode until there are no bytes left in the output buffer. Then we go
      // back to encoding/flushing.
      if (draining) {
        totalBytesRead += drain(b, off + totalBytesRead, len - totalBytesRead);
        if (totalBytesRead == len || doneFlushing) {

        if (existing != null) {
          return existing;
        }
      }
      drainQueue();
      return created;
    }

    // N.B. Draining the queue is only necessary to ensure that we don't accumulate empty references
    // in the array. We could skip this if we decide we don't care about holding on to Reference
    // objects indefinitely.

func (c *Connection) reconnected() {
	c.updateState(StateConnectionError)
	c.reconnects.Add(1)

	// Drain the outQueue, so any blocked messages can be sent.
	// We keep the queue, but start draining it, if it gets full.
	stopDraining := make(chan struct{})
	var wg sync.WaitGroup
	wg.Add(1)
	defer func() {
		close(stopDraining)
		wg.Wait()
	}()
	go func() {
		defer wg.Done()
		for {

- [Kube-proxy]: Implemented connection draining for terminating nodes. ([#116470](https://github.com/kubernetes/kubernetes/pull/116470), [@alexanderConstantinescu](https://github.com/alexanderConstantinescu))

  public void testIsPresent_yes() {
    assertTrue(Optional.of("training").isPresent());
  }

  public void testGet_absent() {
    Optional<String> optional = Optional.absent();
    assertThrows(IllegalStateException.class, optional::get);
  }

  public void testGet_present() {
    assertThat(Optional.of("training").get()).isEqualTo("training");
  }

     * @return this ExecJob instance for method chaining
     */
    public ExecJob jobExecutor(final JobExecutor jobExecutor) {
        this.jobExecutor = jobExecutor;
        return this;
    }

    /**
     * Sets the session ID for this job execution.
     *
     * @param sessionId the unique session identifier
     * @return this ExecJob instance for method chaining
     */

         */
        public SearchResultBuilder addDocument(final Map<String, Object> doc) {
            documentList.add(doc);
            return this;
        }

        /**
         * Sets the facet response containing aggregated facet information.
         *
         * @param facetResponse The facet response
         * @return This builder instance for method chaining
         */

     *
     * @return create cancel request
     */
    CommonServerMessageBlockRequest createCancel();

    /**
     * Checks if chaining is allowed with the next request.
     *
     * @param next the next request in the chain
     * @return whether to allow chaining
     */
    boolean allowChain(CommonServerMessageBlockRequest next);

    /**
     * Sets the tree ID.
     *
     * @param t the tree ID to set

        assertEquals((byte) 0xFF, buf[1], "AndX command should be 0xFF when no chaining");
        assertEquals((byte) 0x00, buf[2], "Reserved byte must be 0");
        assertEquals((byte) 0xDE, buf[3], "Sentinel low offset when no chaining");
        assertEquals((byte) 0xDE, buf[4], "Sentinel high offset when no chaining");
        assertNull(block.andx, "andx should be cleared when not chaining");
    }

    @Test

        return type;
    }

    /**
     * Sets the highlighting type with fluent interface.
     *
     * @param type the highlighting type to set
     * @return this HighlightInfo instance for method chaining
     */
    public HighlightInfo type(final String type) {
        this.type = type;
        return this;
    }

    /**
     * Gets the fragment size.
     *