func (s *DiscoveryServer) processRequest(req *discovery.DiscoveryRequest, con *Connection) error {
	stype := v3.GetShortType(req.TypeUrl)
	log.Debugf("ADS:%s: REQ %s resources:%d nonce:%s version:%s ", stype,
		con.ID(), len(req.ResourceNames), req.ResponseNonce, req.VersionInfo)
	if req.TypeUrl == v3.HealthInfoType {
		s.handleWorkloadHealthcheck(con.proxy, req)
		return nil

	s := xds.NewFakeDiscoveryServer(t, xds.FakeOptions{})
	ads := s.ConnectDeltaADS().WithType(v3.ClusterType)
	ads.RequestResponseAck(nil)
}

func TestDeltaAdsClusterUpdate(t *testing.T) {
	s := xds.NewFakeDiscoveryServer(t, xds.FakeOptions{})
	ads := s.ConnectDeltaADS().WithType(v3.EndpointType)
	nonce := ""
	sendEDSReqAndVerify := func(add, remove, expect []string) {
		t.Helper()
		res := ads.RequestResponseAck(&discovery.DeltaDiscoveryRequest{

	s := xdsfake.NewFakeDiscoveryServer(t, xdsfake.FakeOptions{})

	ads := s.ConnectADS().WithType(v3.EndpointType)
	res := ads.RequestResponseAck(t, &discovery.DiscoveryRequest{ResourceNames: []string{"fake-cluster"}})
	// Close the connection and reconnect
	ads.Cleanup()

	ads = s.ConnectADS().WithType(v3.EndpointType)

	// Reconnect with the same resources
	ads.RequestResponseAck(t, &discovery.DiscoveryRequest{

		})
		ads := s.ConnectDeltaADS().WithType(v3.AddressType).WithMetadata(model.NodeMetadata{NodeName: "node"})
		ads.Request(&discovery.DeltaDiscoveryRequest{
			ResourceNamesSubscribe:   []string{"*"},
			ResourceNamesUnsubscribe: []string{"*"},
		})
		ads.ExpectEmptyResponse()

		// Now subscribe to the pod, should get it back
		resp := ads.RequestResponseAck(&discovery.DeltaDiscoveryRequest{

		})
		if _, err := ads.Wait(time.Second*5, watchAll...); err != nil {
			t.Fatal(err)
		}
		if len(ads.GetEndpoints()) != 1 {
			t.Fatalf("Expected a partial EDS update, but got: %v", xdstest.MapKeys(ads.GetEndpoints()))
		}
	})
	t.Run("Full Push with updated services and virtual services", func(t *testing.T) {
		ads.WaitClear()
		s.Discovery.Push(&model.PushRequest{
			Full: true,

	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
	}
	log.Debugf("ADS:%s: RESOURCE CHANGE added %v removed %v %s %s %s", stype,

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

		deltaLog.Debugf("ADS:%s: ACK %s %s", stype, con.ID(), request.ResponseNonce)
		return false
	}
	deltaLog.Debugf("ADS:%s: RESOURCE CHANGE previous resources: %v, new resources: %v %s %s", stype,

	t.Helper()
	ads := s.ConnectADS().WithID(sidecarID)

	metadata := &structpb.Struct{Fields: map[string]*structpb.Value{
		"ISTIO_VERSION": {Kind: &structpb.Value_StringValue{StringValue: "1.3"}},
		"NETWORK":       {Kind: &structpb.Value_StringValue{StringValue: network}},
	}}

	ads.RequestResponseAck(t, &discovery.DiscoveryRequest{
		Node: &core.Node{

	ClusterDiscoveryType: &cluster.Cluster_Type{Type: cluster.Cluster_EDS},
	EdsClusterConfig: &cluster.Cluster_EdsClusterConfig{
		EdsConfig: &core.ConfigSource{
			ConfigSourceSpecifier: &core.ConfigSource_Ads{
				Ads: &core.AggregatedConfigSource{},
			},
		},
	},
	LbPolicy: cluster.Cluster_ROUND_ROBIN,
	TransportSocket: &core.TransportSocket{
		Name: wellknown.TransportSocketTLS,
		ConfigType: &core.TransportSocket_TypedConfig{

  // the merged pad will not change anything, while the un-merged will remove a
  // value, then insert a 0 at its place. This only holds for low and high pads,
  // the spec does not allow negative interior pads, so we don't check there.
  auto low_pads = pad_op.getEdgePaddingLow().getValues<IntegerAttr>();
  auto parent_low_pads =
      parent_pad.getEdgePaddingLow().getValues<IntegerAttr>();