// even though Nonce match and it looks like an ACK.
	if alwaysRespond {
		log.Infof("ADS:%s: FORCE RESPONSE %s for warming.", stype, id)
		return true, emptyResourceDelta
	}

	if len(removed) == 0 && len(added) == 0 {
		log.Debugf("ADS:%s: ACK %s %s %s", stype, id, request.VersionInfo, request.ResponseNonce)
		return false, emptyResourceDelta
	}

[2020-01-01 00:00:00] >> 0x00000003    47 HEADERS       END_STREAM|END_HEADERS
[2020-01-01 00:00:00] << 0x00000000     6 SETTINGS
[2020-01-01 00:00:00] << 0x00000000     0 SETTINGS      ACK
[2020-01-01 00:00:00] << 0x00000000     4 WINDOW_UPDATE
[2020-01-01 00:00:00] >> 0x00000000     0 SETTINGS      ACK
[2020-01-01 00:00:00] << 0x00000003   322 HEADERS       END_HEADERS
[2020-01-01 00:00:00] << 0x00000003   288 DATA

	}
	// Acknowledge Mux created.
	// Send async.
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		var ack message
		ack.Op = OpAckMux
		ack.Flags = m.BaseFlags
		ack.MuxID = m.ID
		m.send(ack)
		if debugPrint {
			fmt.Println("connected stream mux:", ack.MuxID)
		}
	}()

	// Data inbound to the handler
	var handlerIn chan []byte
	if inboundCap > 0 {

	if ok {
		return Key[O](keyFunc(ac.Name, ac.Namespace)), true
	}
	arn, ok := any(a).(ResourceNamer)
	if ok {
		return Key[O](arn.ResourceName()), true
	}
	ack := GetApplyConfigKey(a)
	if ack != nil {
		return *ack, true
	}
	return "", false
}

// getLabels returns the labels for an object, if possible.
// Warning: this will panic if the labels is not available.

      }
      sink.flush()
    }
  }

  /**
   * Send a connection-level ping to the peer. `ack` indicates this is a reply. The data in
   * `payload1` and `payload2` opaque binary, and there are no rules on the content.
   */
  @Throws(IOException::class)
  fun ping(
    ack: Boolean,
    payload1: Int,
    payload2: Int,
  ) {
    this.withLock {
      if (closed) throw IOException("closed")

		xds.ExpiredNonce.With(typeTag.Value(v3.GetMetricType(request.TypeUrl))).Increment()
		return false
	}
	// If it comes here, that means nonce match. This an ACK. We should record
	// the ack details and respond if there is a change in resource names.
	var previousResources, currentResources []string
	var alwaysRespond bool

		},
	})
	if err != nil {
		t.Fatal(err)
	}

	// Wait until nonce was updated, which represents an ACK/NACK has been received.
	retry.UntilSuccessOrFail(t, func() error {
		if proxy.ecdsLastNonce.Load() == n1.Load() {
			return errors.New("last process nonce has not been updated. no ecds ack/nack is received yet")
		}
		return nil
	}, retry.Timeout(time.Second), retry.Delay(time.Millisecond))

                try {
                    tree.send(create2);
                }
                catch ( Exception e ) {
                    // timeout is expected for now as we do not ack the break
                    if ( ! ( e.getCause() instanceof RequestTimeoutException ) ) {
                        throw e;
                    }
                }
            }

	// Wait until nonce was updated, which represents an ACK/NACK has been received.
	retry.UntilSuccessOrFail(t, func() error {
		if proxy.ecdsLastNonce.Load() == n1.Load() {
			return errors.New("last process nonce has not been updated. no ecds ack/nack is received yet")
		}
		return nil
	}, retry.Timeout(time.Second), retry.Delay(time.Millisecond))

// EventHandler allows for generic monitoring of xDS ACKS and disconnects, for the purpose of tracking
// Config distribution through the mesh.
type DistributionStatusCache interface {
	// RegisterEvent notifies the implementer of an xDS ACK, and must be non-blocking
	RegisterEvent(conID string, eventType EventType, nonce string)
	RegisterDisconnect(s string, types sets.Set[EventType])
	QueryLastNonce(conID string, eventType EventType) (noncePrefix string)