t.Helper()
	req = a.fillInRequestDefaults(req)
	if err := a.client.Send(req); err != nil {
		t.Fatal(err)
	}
}

// RequestResponseAck does a full XDS exchange: Send a request, get a response, and ACK the response
func (a *AdsTest) RequestResponseAck(t test.Failer, req *discovery.DiscoveryRequest) *discovery.DiscoveryResponse {
	t.Helper()
	req = a.fillInRequestDefaults(req)
	a.Request(t, req)

		info = " " + logdata.AdditionalInfo
	}
	if len(logFiltered) > 0 {
		info += logFiltered
	}
	if err != nil || res == nil {
		// If we have nothing to send, report that we got an ACK for this version.
		if s.StatusReporter != nil {
			s.StatusReporter.RegisterEvent(con.ID(), w.TypeUrl, req.Push.LedgerVersion)
		}
		if log.DebugEnabled() {

	req = a.fillInRequestDefaults(req)
	if err := a.client.Send(req); err != nil {
		a.t.Fatal(err)
	}
}

// RequestResponseAck does a full XDS exchange: Send a request, get a response, and ACK the response
func (a *DeltaAdsTest) RequestResponseAck(req *discovery.DeltaDiscoveryRequest) *discovery.DeltaDiscoveryResponse {
	a.t.Helper()
	req = a.fillInRequestDefaults(req)
	a.Request(req)
	resp := a.ExpectResponse()

	"istio.io/istio/pilot/pkg/model"
	"istio.io/istio/pilot/pkg/xds"
	v3 "istio.io/istio/pilot/pkg/xds/v3"
	xdsfake "istio.io/istio/pilot/test/xds"
)

func TestSyncz(t *testing.T) {
	t.Run("return the sent and ack status of adsClient connections", func(t *testing.T) {
		s := xdsfake.NewFakeDiscoveryServer(t, xdsfake.FakeOptions{})
		ads := s.ConnectADS()

		ads.RequestResponseAck(t, &discovery.DiscoveryRequest{TypeUrl: v3.ClusterType})

			Body:         data,
		}); err != nil {
		return err
	}

	// check for publisher confirms only if its enabled
	if target.args.PublisherConfirms {
		confirmed := <-confirms
		if !confirmed.Ack {
			return fmt.Errorf("failed delivery of delivery tag: %d", confirmed.DeliveryTag)
		}
	}

	return nil
}

// Save - saves the events to the store which will be replayed when the amqp connection is active.

	ads.ExpectResponse(t)
	ads.ExpectResponse(t)
	// Close the connection and reconnect
	ads.Cleanup()
	ads = s.ConnectADS()

	// Send a request for EDS version 1 - we do not explicitly ACK this.
	ads.Request(t, &discovery.DiscoveryRequest{
		TypeUrl:       v3.EndpointType,
		ResourceNames: []string{"my-resource"},
		ResponseNonce: eres.Nonce,
	})
	// We should get a response

		Node:          a.node(),
		TypeUrl:       typeurl,
		ResourceNames: rsc,
	})
}

func (a *ADSC) ack(msg *discovery.DiscoveryResponse) {
	var resources []string

	if strings.HasPrefix(msg.TypeUrl, v3.DebugType) {
		// If the response is for istio.io/debug or istio.io/debug/*,
		// skip to send ACK.
		return
	}

	if msg.TypeUrl == v3.EndpointType {
		for c := range a.edsClusters {

				var errorResp *google_rpc.Status
				if err != nil {
					errorResp = &google_rpc.Status{
						Code:    int32(codes.Internal),
						Message: err.Error(),
					}
				}
				// Send ACK/NACK
				con.sendDeltaRequest(&discovery.DeltaDiscoveryRequest{
					TypeUrl:       resp.TypeUrl,
					ResponseNonce: resp.Nonce,
					ErrorDetail:   errorResp,
				})
				continue
			}

	if old, ok := r.status[key]; ok {
		sets.DeleteCleanupLast(r.reverseStatus, old, key)
	}
}

// RegisterDisconnect : when a dataplane disconnects, we should no longer count it, nor expect it to ack config.
func (r *Reporter) RegisterDisconnect(conID string, types sets.Set[xds.EventType]) {
	r.mu.Lock()
	defer r.mu.Unlock()
	for xdsType := range types {
		key := GenStatusReporterMapKey(conID, xdsType)

	assert.Equal(t, resp.RemovedResources, []string{"outbound|80||test-1.default"})

	// Another removal should behave the same
	s.MemRegistry.RemoveService("eds.test.svc.cluster.local")
	resp = ads.ExpectResponse()
	// ACK
	ads.Request(&discovery.DeltaDiscoveryRequest{
		TypeUrl:       resp.TypeUrl,
		ResponseNonce: resp.Nonce,
	})
	assertResources(resp)