diff --git a/staging/src/k8s.io/legacy-cloud-providers/gce/gce_annotations.go b/staging/src/k8s.io/legacy-cloud-providers/gce/gce_annotations.go
index a9cab8f366f..43e11660f78 100644
--- a/staging/src/k8s.io/legacy-cloud-providers/gce/gce_annotations.go
+++ b/staging/src/k8s.io/legacy-cloud-providers/gce/gce_annotations.go
@@ -58,6 +58,9 @@ const (
 
 	// NetworkTierAnnotationPremium is an annotation to indicate the Service is on the Premium network tier
 	NetworkTierAnnotationPremium = cloud.NetworkTierPremium
+
+	// NEGAnnotation is an annotation to indicate that the loadbalancer service is using NEGs instead of InstanceGroups
+	NEGAnnotation = "cloud.google.com/neg"
 )
 
 // GetLoadBalancerAnnotationType returns the type of GCP load balancer which should be assembled.
diff --git a/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_internal.go b/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_internal.go
index 881cf9ab21a..88560a9f961 100644
--- a/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_internal.go
+++ b/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_internal.go
@@ -35,7 +35,16 @@ const (
 	allInstances = "ALL"
 )
 
+func usesNEG(service *v1.Service) bool {
+	_, ok := service.GetAnnotations()[NEGAnnotation]
+	return ok
+}
+
 func (g *Cloud) ensureInternalLoadBalancer(clusterName, clusterID string, svc *v1.Service, existingFwdRule *compute.ForwardingRule, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) {
+	if usesNEG(svc) {
+		// Do not manage loadBalancer for services using NEGs
+		return &svc.Status.LoadBalancer, nil
+	}
 	nm := types.NamespacedName{Name: svc.Name, Namespace: svc.Namespace}
 	ports, protocol := getPortsAndProtocol(svc.Spec.Ports)
 	if protocol != v1.ProtocolTCP && protocol != v1.ProtocolUDP {
@@ -201,6 +210,10 @@ func (g *Cloud) clearPreviousInternalResources(svc *v1.Service, loadBalancerName
 // updateInternalLoadBalancer is called when the list of nodes has changed. Therefore, only the instance groups
 // and possibly the backend service need to be updated.
 func (g *Cloud) updateInternalLoadBalancer(clusterName, clusterID string, svc *v1.Service, nodes []*v1.Node) error {
+	if usesNEG(svc) {
+		// Do not manage loadBalancer for services using NEGs
+		return nil
+	}
 	g.sharedResourceLock.Lock()
 	defer g.sharedResourceLock.Unlock()
 
diff --git a/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_internal_test.go b/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_internal_test.go
index f2aaf029f7e..322f42a9cd8 100644
--- a/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_internal_test.go
+++ b/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_internal_test.go
@@ -847,3 +847,62 @@ func TestCompareHealthChecks(t *testing.T) {
 		})
 	}
 }
+
+func TestEnsureInternalLoadBalancerNEG(t *testing.T) {
+	t.Parallel()
+
+	vals := DefaultTestClusterValues()
+	gce, err := fakeGCECloud(vals)
+	require.NoError(t, err)
+
+	nodeNames := []string{"test-node-1"}
+	nodes, err := createAndInsertNodes(gce, nodeNames, vals.ZoneName)
+	require.NoError(t, err)
+
+	apiService := fakeLoadbalancerServiceWithNEGs(string(LBTypeInternal))
+	status, err := gce.EnsureLoadBalancer(context.Background(), vals.ClusterName, apiService, nodes)
+	assert.NoError(t, err)
+	// No loadbalancer resources will be created due to the NEG annotation
+	assert.Empty(t, status.Ingress)
+	assertInternalLbResourcesDeleted(t, gce, apiService, vals, true)
+	// Invoking delete should be a no-op
+	err = gce.EnsureLoadBalancerDeleted(context.Background(), vals.ClusterName, apiService)
+	assert.NoError(t, err)
+	// Now remove the annotation so that lb resources are created
+	delete(apiService.Annotations, NEGAnnotation)
+	status, err = gce.EnsureLoadBalancer(context.Background(), vals.ClusterName, apiService, nodes)
+	assert.NoError(t, err)
+	assert.NotEmpty(t, status.Ingress)
+	assertInternalLbResources(t, gce, apiService, vals, nodeNames)
+}
+
+func TestEnsureInternalLoadBalancerDeletedNEGs(t *testing.T) {
+	t.Parallel()
+
+	vals := DefaultTestClusterValues()
+	gce, err := fakeGCECloud(vals)
+	require.NoError(t, err)
+
+	nodeNames := []string{"test-node-1"}
+	nodes, err := createAndInsertNodes(gce, nodeNames, vals.ZoneName)
+	require.NoError(t, err)
+
+	apiService := fakeLoadbalancerService(string(LBTypeInternal))
+
+	status, err := gce.EnsureLoadBalancer(context.Background(), vals.ClusterName, apiService, nodes)
+	assert.NoError(t, err)
+	assert.NotEmpty(t, status.Ingress)
+	// Annotation gets added to service
+	apiService.Annotations[NEGAnnotation] = "{\"ilb\": true}"
+	newLBStatus := v1.LoadBalancerStatus{Ingress: []v1.LoadBalancerIngress{{IP: "1.2.3.4"}}}
+	// mock scenario where a different controller modifies status.
+	apiService.Status.LoadBalancer = newLBStatus
+	status, err = gce.EnsureLoadBalancer(context.Background(), vals.ClusterName, apiService, nodes)
+	assert.NoError(t, err)
+	// ensure that the status info is intact
+	assert.Equal(t, status, &newLBStatus)
+	// Invoked when service is deleted.
+	err = gce.EnsureLoadBalancerDeleted(context.Background(), vals.ClusterName, apiService)
+	assert.NoError(t, err)
+	assertInternalLbResourcesDeleted(t, gce, apiService, vals, true)
+}
diff --git a/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_utils_test.go b/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_utils_test.go
index 87fcf1d5dcc..6dc4c608304 100644
--- a/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_utils_test.go
+++ b/staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_utils_test.go
@@ -62,6 +62,12 @@ func fakeLoadbalancerService(lbType string) *v1.Service {
 	}
 }
 
+func fakeLoadbalancerServiceWithNEGs(lbType string) *v1.Service {
+	svc := fakeLoadbalancerService(lbType)
+	svc.Annotations[NEGAnnotation] = "{\"ilb\": true}"
+	return svc
+}
+
 var (
 	FilewallChangeMsg = fmt.Sprintf("%s %s %s", v1.EventTypeNormal, eventReasonManualChange, eventMsgFirewallChange)
 )