Repository: kubernetes-sigs/kueue

Issue:
MVP support for serving workloads running as LeaderWorkerSet
**What would you like to be added**:

MVP support for LeaderWorkerSet in Kueue. It does not need to be ideal, but we want to have some support to unblock users and collect users' feedback. 

The idea is to base the support on StatefulSets, so the integration would also use Pod Groups, similarly as for regular StatefulSets. Each LeaderWorkerGroup creates a new Pod Group. I a single pod group we will have:

- Leader pod, controller by Leader’s STS
- Worker pods, controller by unique, dedicated STS

The size of the group will be taken from LeaderWorkerSet.Spec.LeaderWorkerTemplate.Size and increased by 1 (to include the leader).

This is a follow up to https://github.com/kubernetes-sigs/kueue/issues/2717.

**Why is this needed**:

We want to support serving primitives in Kueue as there is an increasing demand among users to run clusters mixing AI training and inference who want to manage the expensive GPU resources.

[LeaderWorkerSet](https://github.com/kubernetes-sigs/lws) is a new serving API which is gaining popularity as a primitive to host AI/ML inference. 


Discussion:
@mimowo: /assign @vladikkuzn @mimowo: /cc @mwielgus @tenzen-y @mimowo: I synced with @mbobrovskyi and @vladikkuzn on the feature and it seems complex, so I propose to have a KEP for it, and go via the Alpha phase so that we can update the implementation in the future easily.@mimowo: The already identified follow ups needed after https://github.com/kubernetes-sigs/kueue/pull/3515:
- support for resizing the LWS group by group  (potentially the same PR, but might be follow up) [already done in the main PR]
- support for the default StartupPolicy=LeaderReady
- support for TAS with rank-based ordering (with e2e tests)
- https://github.com/kubernetes-sigs/kueue/issues/4000@mimowo: /reopen
to finalize the integration with TAS @k8s-ci-robot: @mimowo: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/3232#issuecomment-2619313252):

>/reopen
>to finalize the integration with TAS 


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@mimowo: Also, I would like to have e2e test for scaling the LWS when Startup policy is Leader ready (default). @mbobrovskyi: /unassign @vladikkuzn 
/assign@tenzen-y: @mbobrovskyi @mimowo Will you finalize MultiKueue and TAS support within this MVP issue?
@kannon92: What outstanding items do we have for this ticket?@mimowo: I think we can close it, AFAIK the only outstanding work for now is the support of rank-based ordering in TAS for groups.
However, I would  not call it MVP any longer: https://github.com/kubernetes-sigs/kueue/issues/4531@mimowo: /close
Thank you @mbobrovskyi for the implementation!@k8s-ci-robot: @mimowo: Closing this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/3232#issuecomment-2709742137):

>/close
>Thank you @mbobrovskyi for the implementation!


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>

Solution patch:
diff --git a/Makefile-test.mk b/Makefile-test.mk
index 8e82cadcd..f51cb33bd 100644
--- a/Makefile-test.mk
+++ b/Makefile-test.mk
@@ -142,6 +142,7 @@ run-test-multikueue-e2e-%: FORCE
 	@echo Running multikueue e2e for k8s ${K8S_VERSION}
 	E2E_KIND_VERSION="kindest/node:v$(K8S_VERSION)" KIND_CLUSTER_NAME=$(KIND_CLUSTER_NAME) CREATE_KIND_CLUSTER=$(CREATE_KIND_CLUSTER) \
 		ARTIFACTS="$(ARTIFACTS)/$@" IMAGE_TAG=$(IMAGE_TAG) GINKGO_ARGS="$(GINKGO_ARGS)" \
+		APPWRAPPER_VERSION=$(APPWRAPPER_VERSION) \
 		JOBSET_VERSION=$(JOBSET_VERSION) KUBEFLOW_VERSION=$(KUBEFLOW_VERSION) \
 		KUBEFLOW_MPI_VERSION=$(KUBEFLOW_MPI_VERSION) KUBERAY_VERSION=$(KUBERAY_VERSION) \
 		./hack/multikueue-e2e-test.sh
@@ -153,6 +154,7 @@ run-test-tas-e2e-%: FORCE
 	E2E_KIND_VERSION="kindest/node:v$(K8S_VERSION)" KIND_CLUSTER_NAME=$(KIND_CLUSTER_NAME) CREATE_KIND_CLUSTER=$(CREATE_KIND_CLUSTER) \
 		ARTIFACTS="$(ARTIFACTS)/$@" IMAGE_TAG=$(IMAGE_TAG) GINKGO_ARGS="$(GINKGO_ARGS)" \
 		JOBSET_VERSION=$(JOBSET_VERSION) KUBEFLOW_VERSION=$(KUBEFLOW_VERSION) KUBEFLOW_MPI_VERSION=$(KUBEFLOW_MPI_VERSION) \
+		APPWRAPPER_VERSION=$(APPWRAPPER_VERSION) \
 		KIND_CLUSTER_FILE="tas-kind-cluster.yaml" E2E_TARGET_FOLDER="tas" \
 		./hack/e2e-test.sh
 	$(PROJECT_DIR)/bin/ginkgo-top -i $(ARTIFACTS)/$@/e2e.json > $(ARTIFACTS)/$@/e2e-top.yaml
diff --git a/hack/multikueue-e2e-test.sh b/hack/multikueue-e2e-test.sh
index 5fa44f50d..d143199e5 100755
--- a/hack/multikueue-e2e-test.sh
+++ b/hack/multikueue-e2e-test.sh
@@ -81,6 +81,11 @@ function kind_load {
     install_jobset "$WORKER1_KIND_CLUSTER_NAME"
     install_jobset "$WORKER2_KIND_CLUSTER_NAME"
 
+    # APPWRAPPER SETUP
+    install_appwrapper "$MANAGER_KIND_CLUSTER_NAME"
+    install_appwrapper "$WORKER1_KIND_CLUSTER_NAME"
+    install_appwrapper "$WORKER2_KIND_CLUSTER_NAME"
+
     # KUBEFLOW SETUP
     # In order for MPI-operator and Training-operator to work on the same cluster it is required that:
     # 1. 'kubeflow.org_mpijobs.yaml' is removed from base/crds/kustomization.yaml - https://github.com/kubeflow/training-operator/issues/1930
diff --git a/pkg/controller/jobframework/validation.go b/pkg/controller/jobframework/validation.go
index 3e37f7229..40fdc2f9a 100644
--- a/pkg/controller/jobframework/validation.go
+++ b/pkg/controller/jobframework/validation.go
@@ -25,6 +25,7 @@ import (
 
 	kfmpi "github.com/kubeflow/mpi-operator/pkg/apis/kubeflow/v2beta1"
 	kftraining "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1"
+	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -56,7 +57,9 @@ var (
 		kfmpi.SchemeGroupVersion.WithKind(kfmpi.Kind).String(),
 		rayv1.SchemeGroupVersion.WithKind("RayJob").String(),
 		corev1.SchemeGroupVersion.WithKind("Pod").String(),
-		rayv1.SchemeGroupVersion.WithKind("RayCluster").String())
+		rayv1.SchemeGroupVersion.WithKind("RayCluster").String(),
+		awv1beta2.GroupVersion.WithKind("AppWrapper").String(),
+	)
 )
 
 // ValidateJobOnCreate encapsulates all GenericJob validations that must be performed on a Create operation
diff --git a/pkg/controller/jobs/appwrapper/appwrapper_controller.go b/pkg/controller/jobs/appwrapper/appwrapper_controller.go
index ccc4c0924..907f1eb17 100644
--- a/pkg/controller/jobs/appwrapper/appwrapper_controller.go
+++ b/pkg/controller/jobs/appwrapper/appwrapper_controller.go
@@ -33,6 +33,7 @@ import (
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/podset"
 )
 
@@ -61,6 +62,7 @@ func init() {
 		SetupIndexes:           SetupIndexes,
 		AddToScheme:            awv1beta2.AddToScheme,
 		IsManagingObjectsOwner: isAppWrapper,
+		MultiKueueAdapter:      &multiKueueAdapter{},
 	}))
 }
 
@@ -91,6 +93,7 @@ func SetupIndexes(ctx context.Context, indexer client.FieldIndexer) error {
 type AppWrapper awv1beta2.AppWrapper
 
 var _ jobframework.GenericJob = (*AppWrapper)(nil)
+var _ jobframework.JobWithManagedBy = (*AppWrapper)(nil)
 
 func fromObject(o runtime.Object) *AppWrapper {
 	return (*AppWrapper)(o.(*awv1beta2.AppWrapper))
@@ -122,6 +125,9 @@ func (aw *AppWrapper) PodSets() ([]kueue.PodSet, error) {
 		ctrl.Log.Error(err, "Error returned from awutils.GetPodSets", "appwrapper", aw)
 		return nil, err
 	}
+	for idx := range podSets {
+		podSets[idx].TopologyRequest = jobframework.PodSetTopologyRequest(&podSets[idx].Template.ObjectMeta, nil, nil, nil)
+	}
 	return podSets, nil
 }
 
@@ -156,6 +162,20 @@ func (aw *AppWrapper) PodsReady() bool {
 	return meta.IsStatusConditionTrue(aw.Status.Conditions, string(awv1beta2.PodsReady))
 }
 
+func (j *AppWrapper) CanDefaultManagedBy() bool {
+	jobSpecManagedBy := j.Spec.ManagedBy
+	return features.Enabled(features.MultiKueue) &&
+		(jobSpecManagedBy == nil || *jobSpecManagedBy == awv1beta2.AppWrapperControllerName)
+}
+
+func (j *AppWrapper) ManagedBy() *string {
+	return j.Spec.ManagedBy
+}
+
+func (j *AppWrapper) SetManagedBy(managedBy *string) {
+	j.Spec.ManagedBy = managedBy
+}
+
 func GetWorkloadNameForAppWrapper(jobName string, jobUID types.UID) string {
 	return jobframework.GetWorkloadNameForOwnerWithGVK(jobName, jobUID, gvk)
 }
diff --git a/pkg/controller/jobs/appwrapper/appwrapper_multikueue_adapter.go b/pkg/controller/jobs/appwrapper/appwrapper_multikueue_adapter.go
new file mode 100644
index 000000000..d63cce42d
--- /dev/null
+++ b/pkg/controller/jobs/appwrapper/appwrapper_multikueue_adapter.go
@@ -0,0 +1,137 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package appwrapper
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/klog/v2"
+	"k8s.io/utils/ptr"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/controller/constants"
+	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+	"sigs.k8s.io/kueue/pkg/util/api"
+	clientutil "sigs.k8s.io/kueue/pkg/util/client"
+)
+
+type multiKueueAdapter struct{}
+
+var _ jobframework.MultiKueueAdapter = (*multiKueueAdapter)(nil)
+
+func (b *multiKueueAdapter) SyncJob(ctx context.Context, localClient client.Client, remoteClient client.Client, key types.NamespacedName, workloadName, origin string) error {
+	log := ctrl.LoggerFrom(ctx)
+
+	localAppWrapper := awv1beta2.AppWrapper{}
+	err := localClient.Get(ctx, key, &localAppWrapper)
+	if err != nil {
+		return err
+	}
+
+	remoteAppWrapper := awv1beta2.AppWrapper{}
+	err = remoteClient.Get(ctx, key, &remoteAppWrapper)
+	if client.IgnoreNotFound(err) != nil {
+		return err
+	}
+
+	// if the remote exists, just copy the status
+	if err == nil {
+		if localAppWrapper.Spec.Suspend {
+			// Ensure the appwrapper is unsuspended before updating its status; otherwise, it will fail when patching the spec.
+			log.V(2).Info("Skipping the sync since the local appwrapper is still suspended")
+			return nil
+		}
+		return clientutil.PatchStatus(ctx, localClient, &localAppWrapper, func() (bool, error) {
+			localAppWrapper.Status = remoteAppWrapper.Status
+			return true, nil
+		})
+	}
+
+	// Make a copy of the local AppWrapper
+	remoteAppWrapper = awv1beta2.AppWrapper{
+		ObjectMeta: api.CloneObjectMetaForCreation(&localAppWrapper.ObjectMeta),
+		Spec:       *localAppWrapper.Spec.DeepCopy(),
+	}
+
+	// add the prebuilt workload
+	if remoteAppWrapper.Labels == nil {
+		remoteAppWrapper.Labels = map[string]string{}
+	}
+	remoteAppWrapper.Labels[constants.PrebuiltWorkloadLabel] = workloadName
+	remoteAppWrapper.Labels[kueue.MultiKueueOriginLabel] = origin
+
+	// clear the managedBy to enable the remote AppWrapper controller to take over
+	remoteAppWrapper.Spec.ManagedBy = nil
+
+	return remoteClient.Create(ctx, &remoteAppWrapper)
+}
+
+func (b *multiKueueAdapter) DeleteRemoteObject(ctx context.Context, remoteClient client.Client, key types.NamespacedName) error {
+	aw := &awv1beta2.AppWrapper{}
+	aw.SetName(key.Name)
+	aw.SetNamespace(key.Namespace)
+	return client.IgnoreNotFound(remoteClient.Delete(ctx, aw))
+}
+
+func (b *multiKueueAdapter) GVK() schema.GroupVersionKind {
+	return gvk
+}
+
+func (b *multiKueueAdapter) KeepAdmissionCheckPending() bool {
+	return false
+}
+
+func (b *multiKueueAdapter) IsJobManagedByKueue(ctx context.Context, c client.Client, key types.NamespacedName) (bool, string, error) {
+	aw := awv1beta2.AppWrapper{}
+	err := c.Get(ctx, key, &aw)
+	if err != nil {
+		return false, "", err
+	}
+	awControllerName := ptr.Deref(aw.Spec.ManagedBy, "")
+	if awControllerName != kueue.MultiKueueControllerName {
+		return false, fmt.Sprintf("Expecting spec.managedBy to be %q not %q", kueue.MultiKueueControllerName, awControllerName), nil
+	}
+	return true, "", nil
+}
+
+var _ jobframework.MultiKueueWatcher = (*multiKueueAdapter)(nil)
+
+func (*multiKueueAdapter) GetEmptyList() client.ObjectList {
+	return &awv1beta2.AppWrapperList{}
+}
+
+func (*multiKueueAdapter) WorkloadKeyFor(o runtime.Object) (types.NamespacedName, error) {
+	aw, ok := o.(*awv1beta2.AppWrapper)
+	if !ok {
+		return types.NamespacedName{}, errors.New("not an appwrapper")
+	}
+
+	prebuiltWl, hasPrebuiltWorkload := aw.Labels[constants.PrebuiltWorkloadLabel]
+	if !hasPrebuiltWorkload {
+		return types.NamespacedName{}, fmt.Errorf("no prebuilt workload found for appwrapper: %s", klog.KObj(aw))
+	}
+
+	return types.NamespacedName{Name: prebuiltWl, Namespace: aw.Namespace}, nil
+}
diff --git a/site/content/en/docs/overview/_index.md b/site/content/en/docs/overview/_index.md
index c1909610b..cdbc5269e 100644
--- a/site/content/en/docs/overview/_index.md
+++ b/site/content/en/docs/overview/_index.md
@@ -41,7 +41,7 @@ A core design principle for Kueue is to avoid duplicating mature functionality i
 | Feature                                                                                                         | Batch&nbsp;Job | JobSet | PaddleJob | PytorchJob | TFJob | XGBoostJob | MPIJob | Pod | RayCluster | RayJob | AppWrapper |
 |-----------------------------------------------------------------------------------------------------------------|----------------|--------|-----------|------------|-------|------------|--------|-----|------------|--------|------------|
 | [Dynamic Reclaim](/docs/concepts/workload/#dynamic-reclaim)                                                     | +              | +      |           |            |       |            |        | +   |            |        |            |
-| [MultiKueue](/docs/concepts/multikueue/)                                                                        | +              | +      | +         | +          | +     | +          | +      |     | +          | +      |            |
+| [MultiKueue](/docs/concepts/multikueue/)                                                                        | +              | +      | +         | +          | +     | +          | +      |     | +          | +      | +          |
 | [MultiKueueBatchJobWithManagedBy](/docs/concepts/multikueue/#multikueuebatchjobwithmanagedby-enabled)           | +              |        |           |            |       |            |        |     |            |        |            |
 | [PartialAdmission](/docs/tasks/run/jobs/#partial-admission)                                                     | +              |        |           |            |       |            |        |     |            |        |            |
 | [Workload Priority Class](/docs/concepts/workload_priority_class/)                                              | +              | +      | +         | +          | +     | +          | +      | +   | +          | +      | +          |
@@ -53,7 +53,7 @@ A core design principle for Kueue is to avoid duplicating mature functionality i
 | [LendingLimit](/docs/concepts/cluster_queue/#lendinglimit)                                                      | +              | +      | +         | +          | +     | +          | +      | +   | +          | +      | +          |
 | [All-or-nothing with ready Pods](/docs/concepts/workload/#all-or-nothing-semantics-for-job-resource-assignment) | +              | +      | +         | +          | +     | +          | +      | +   | +          | +      | +          |
 | [Fair Sharing](/docs/concepts/preemption/#fair-sharing)                                                         | +              | +      | +         | +          | +     | +          | +      | +   | +          | +      | +          |
-| [Topology Aware Scheduling](/docs/concepts/topology_aware_scheduling)                                           | +              | +      | +         | +          | +     | +          | +      | +   | +          | +      |            |
+| [Topology Aware Scheduling](/docs/concepts/topology_aware_scheduling)                                           | +              | +      | +         | +          | +     | +          | +      | +   | +          | +      | +          |
 
 ## High-level Kueue operation
 
diff --git a/site/content/en/docs/tasks/run/appwrappers.md b/site/content/en/docs/tasks/run/appwrappers.md
index 2ee49285f..fc1c0cae6 100644
--- a/site/content/en/docs/tasks/run/appwrappers.md
+++ b/site/content/en/docs/tasks/run/appwrappers.md
@@ -21,7 +21,7 @@ This guide is for [batch users](/docs/tasks#batch-user) that have a basic unders
 
 ## Before you begin
 
-1. Make sure you are using Kueue v0.11.0 version or newer and AppWrapper v1.0.0 or newer.
+1. Make sure you are using Kueue v0.11.0 version or newer and AppWrapper v1.0.2 or newer.
 
 2. Check [Administer cluster quotas](/docs/tasks/manage/administer_cluster_quotas) for details on the initial Kueue setup.
 
@@ -29,7 +29,7 @@ This guide is for [batch users](/docs/tasks#batch-user) that have a basic unders
 Standalone configuration of the AppWrapper controller. This disables the AppWrapper controller's instance of Kueue's
 GenericJob Reconciller. One way to do this is by doing
 ```
-kustomize build "https://github.com/project-codeflare/appwrapper/config/standalone?ref=v1.0.0"
+kustomize build "https://github.com/project-codeflare/appwrapper/config/standalone?ref=v1.0.2"
 ```
 A future release of AppWrapper will change its default configuration to disable its copy of the GenericJob Reconciller.
 
diff --git a/site/content/en/docs/tasks/run/external_workloads/wrapped_custom_workload.md b/site/content/en/docs/tasks/run/external_workloads/wrapped_custom_workload.md
index 60390f5fc..015e4c019 100644
--- a/site/content/en/docs/tasks/run/external_workloads/wrapped_custom_workload.md
+++ b/site/content/en/docs/tasks/run/external_workloads/wrapped_custom_workload.md
@@ -17,7 +17,7 @@ This guide is for [batch users](/docs/tasks#batch-user) that have a basic unders
 
 ## Before you begin
 
-1. Make sure you are using Kueue v0.11.0 version or newer and AppWrapper v1.0.0 or newer.
+1. Make sure you are using Kueue v0.11.0 version or newer and AppWrapper v1.0.2 or newer.
 
 2. Follow the steps in [Run AppWrappers](/docs/tasks/run/appwrappers/#before-you-begin)
 to learn how to enable and configure the `workload.codeflare.dev/appwrapper` integration.



Test patch:
diff --git a/pkg/controller/jobs/appwrapper/appwrapper_multikueue_adapter_test.go b/pkg/controller/jobs/appwrapper/appwrapper_multikueue_adapter_test.go
new file mode 100644
index 000000000..2ced4ed02
--- /dev/null
+++ b/pkg/controller/jobs/appwrapper/appwrapper_multikueue_adapter_test.go
@@ -0,0 +1,231 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package appwrapper
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/controller/constants"
+	"sigs.k8s.io/kueue/pkg/util/slices"
+	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+	utiltestingaw "sigs.k8s.io/kueue/pkg/util/testingjobs/appwrapper"
+	utiltestingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
+)
+
+const (
+	TestNamespace = "ns"
+)
+
+func TestMultiKueueAdapter(t *testing.T) {
+	objCheckOpts := []cmp.Option{
+		cmpopts.IgnoreFields(metav1.ObjectMeta{}, "ResourceVersion"),
+		cmpopts.IgnoreFields(awv1beta2.AppWrapperComponent{}, "Template"),
+		cmpopts.EquateEmpty(),
+	}
+
+	baseAppWrapperBuilder := utiltestingaw.MakeAppWrapper("aw1", TestNamespace).
+		Component(utiltestingjob.MakeJob("job", "ns").SetTypeMeta().Parallelism(2).Obj())
+	baseAppWrapperManagedByKueueBuilder := baseAppWrapperBuilder.DeepCopy().ManagedBy(kueue.MultiKueueControllerName)
+
+	cases := map[string]struct {
+		managersAppWrappers []awv1beta2.AppWrapper
+		workerAppWrappers   []awv1beta2.AppWrapper
+
+		operation func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error
+
+		wantError               error
+		wantManagersAppWrappers []awv1beta2.AppWrapper
+		wantWorkerAppWrappers   []awv1beta2.AppWrapper
+	}{
+
+		"sync creates missing remote appwrapper": {
+			managersAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperManagedByKueueBuilder.DeepCopy().Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				return adapter.SyncJob(ctx, managerClient, workerClient, types.NamespacedName{Name: "aw1", Namespace: TestNamespace}, "wl1", "origin1")
+			},
+
+			wantManagersAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperManagedByKueueBuilder.DeepCopy().Obj(),
+			},
+			wantWorkerAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperBuilder.DeepCopy().
+					Label(constants.PrebuiltWorkloadLabel, "wl1").
+					Label(kueue.MultiKueueOriginLabel, "origin1").
+					Obj(),
+			},
+		},
+		"sync status from remote appwrapper": {
+			managersAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperManagedByKueueBuilder.DeepCopy().Suspend(false).Obj(),
+			},
+			workerAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperBuilder.DeepCopy().
+					Label(constants.PrebuiltWorkloadLabel, "wl1").
+					Label(kueue.MultiKueueOriginLabel, "origin1").
+					SetPhase(awv1beta2.AppWrapperSuspended).
+					Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				return adapter.SyncJob(ctx, managerClient, workerClient, types.NamespacedName{Name: "aw1", Namespace: TestNamespace}, "wl1", "origin1")
+			},
+
+			wantManagersAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperManagedByKueueBuilder.DeepCopy().
+					Suspend(false).
+					SetPhase(awv1beta2.AppWrapperSuspended).
+					Obj(),
+			},
+			wantWorkerAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperBuilder.DeepCopy().
+					Label(constants.PrebuiltWorkloadLabel, "wl1").
+					Label(kueue.MultiKueueOriginLabel, "origin1").
+					SetPhase(awv1beta2.AppWrapperSuspended).
+					Obj(),
+			},
+		},
+		"status is not synced if local appwrapper is still suspended": {
+			managersAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperManagedByKueueBuilder.DeepCopy().Suspend(true).Obj(),
+			},
+			workerAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperBuilder.DeepCopy().
+					Label(constants.PrebuiltWorkloadLabel, "wl1").
+					Label(kueue.MultiKueueOriginLabel, "origin1").
+					SetPhase(awv1beta2.AppWrapperSuspended).
+					Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				return adapter.SyncJob(ctx, managerClient, workerClient, types.NamespacedName{Name: "aw1", Namespace: TestNamespace}, "wl1", "origin1")
+			},
+
+			wantManagersAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperManagedByKueueBuilder.DeepCopy().
+					Suspend(true).
+					SetPhase(awv1beta2.AppWrapperEmpty).
+					Obj(),
+			},
+			wantWorkerAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperBuilder.DeepCopy().
+					Label(constants.PrebuiltWorkloadLabel, "wl1").
+					Label(kueue.MultiKueueOriginLabel, "origin1").
+					SetPhase(awv1beta2.AppWrapperSuspended).
+					Obj(),
+			},
+		},
+		"remote appwrapper is deleted": {
+			workerAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperBuilder.DeepCopy().
+					Label(constants.PrebuiltWorkloadLabel, "wl1").
+					Label(kueue.MultiKueueOriginLabel, "origin1").
+					Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				return adapter.DeleteRemoteObject(ctx, workerClient, types.NamespacedName{Name: "aw1", Namespace: TestNamespace})
+			},
+		},
+		"missing appwrapper is not considered managed": {
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				if isManged, _, _ := adapter.IsJobManagedByKueue(ctx, managerClient, types.NamespacedName{Name: "aw1", Namespace: TestNamespace}); isManged {
+					return errors.New("expecting false")
+				}
+				return nil
+			},
+		},
+		"job with wrong managedBy is not considered managed": {
+			managersAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperBuilder.DeepCopy().Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				if isManged, _, _ := adapter.IsJobManagedByKueue(ctx, managerClient, types.NamespacedName{Name: "aw1", Namespace: TestNamespace}); isManged {
+					return errors.New("expecting false")
+				}
+				return nil
+			},
+			wantManagersAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperBuilder.DeepCopy().Obj(),
+			},
+		},
+
+		"job managedBy multikueue": {
+			managersAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperManagedByKueueBuilder.DeepCopy().Obj(),
+			},
+			operation: func(ctx context.Context, adapter *multiKueueAdapter, managerClient, workerClient client.Client) error {
+				if isManged, _, _ := adapter.IsJobManagedByKueue(ctx, managerClient, types.NamespacedName{Name: "aw1", Namespace: TestNamespace}); !isManged {
+					return errors.New("expecting true")
+				}
+				return nil
+			},
+			wantManagersAppWrappers: []awv1beta2.AppWrapper{
+				*baseAppWrapperManagedByKueueBuilder.DeepCopy().Obj(),
+			},
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			managerBuilder := utiltesting.NewClientBuilder(awv1beta2.AddToScheme).WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge})
+			managerBuilder = managerBuilder.WithLists(&awv1beta2.AppWrapperList{Items: tc.managersAppWrappers})
+			managerBuilder = managerBuilder.WithStatusSubresource(slices.Map(tc.managersAppWrappers, func(w *awv1beta2.AppWrapper) client.Object { return w })...)
+			managerClient := managerBuilder.Build()
+
+			workerBuilder := utiltesting.NewClientBuilder(awv1beta2.AddToScheme).WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge})
+			workerBuilder = workerBuilder.WithLists(&awv1beta2.AppWrapperList{Items: tc.workerAppWrappers})
+			workerClient := workerBuilder.Build()
+
+			ctx, _ := utiltesting.ContextWithLog(t)
+
+			adapter := &multiKueueAdapter{}
+
+			gotErr := tc.operation(ctx, adapter, managerClient, workerClient)
+
+			if diff := cmp.Diff(tc.wantError, gotErr, cmpopts.EquateErrors()); diff != "" {
+				t.Errorf("unexpected error (-want/+got):\n%s", diff)
+			}
+
+			gotManagersAppWrappers := &awv1beta2.AppWrapperList{}
+			if err := managerClient.List(ctx, gotManagersAppWrappers); err != nil {
+				t.Errorf("unexpected list manager's appwrapper error %s", err)
+			} else {
+				if diff := cmp.Diff(tc.wantManagersAppWrappers, gotManagersAppWrappers.Items, objCheckOpts...); diff != "" {
+					t.Errorf("unexpected manager's appwrapper (-want/+got):\n%s", diff)
+				}
+			}
+
+			gotWorkerAppWrappers := &awv1beta2.AppWrapperList{}
+			if err := workerClient.List(ctx, gotWorkerAppWrappers); err != nil {
+				t.Errorf("unexpected list worker's appwrapper error %s", err)
+			} else {
+				if diff := cmp.Diff(tc.wantWorkerAppWrappers, gotWorkerAppWrappers.Items, objCheckOpts...); diff != "" {
+					t.Errorf("unexpected worker's appwrapper (-want/+got):\n%s", diff)
+				}
+			}
+		})
+	}
+}
diff --git a/pkg/util/testingjobs/appwrapper/wrappers.go b/pkg/util/testingjobs/appwrapper/wrappers.go
index 1a5e37b44..5adda05f0 100644
--- a/pkg/util/testingjobs/appwrapper/wrappers.go
+++ b/pkg/util/testingjobs/appwrapper/wrappers.go
@@ -55,6 +55,11 @@ func (aw *AppWrapperWrapper) Obj() *awv1beta2.AppWrapper {
 	return &aw.AppWrapper
 }
 
+// DeepCopy returns a DeepCopy of aw.
+func (aw *AppWrapperWrapper) DeepCopy() *AppWrapperWrapper {
+	return &AppWrapperWrapper{AppWrapper: *aw.AppWrapper.DeepCopy()}
+}
+
 // Label sets a label of the AppWrapper
 func (aw *AppWrapperWrapper) Label(k, v string) *AppWrapperWrapper {
 	if aw.Labels == nil {
@@ -137,3 +142,9 @@ func (aw *AppWrapperWrapper) SetPhase(phase awv1beta2.AppWrapperPhase) *AppWrapp
 	aw.AppWrapper.Status.Phase = phase
 	return aw
 }
+
+// ManagedBy adds a managedby.
+func (aw *AppWrapperWrapper) ManagedBy(c string) *AppWrapperWrapper {
+	aw.AppWrapper.Spec.ManagedBy = &c
+	return aw
+}
diff --git a/test/e2e/multikueue/e2e_test.go b/test/e2e/multikueue/e2e_test.go
index 8b130fc5d..f2f905cd4 100644
--- a/test/e2e/multikueue/e2e_test.go
+++ b/test/e2e/multikueue/e2e_test.go
@@ -28,6 +28,7 @@ import (
 	kftraining "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1"
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
+	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	appsv1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
@@ -41,6 +42,7 @@ import (
 	jobset "sigs.k8s.io/jobset/api/jobset/v1alpha2"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	workloadaw "sigs.k8s.io/kueue/pkg/controller/jobs/appwrapper"
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
 	workloadjobset "sigs.k8s.io/kueue/pkg/controller/jobs/jobset"
 	workloadpytorchjob "sigs.k8s.io/kueue/pkg/controller/jobs/kubeflow/jobs/pytorchjob"
@@ -50,6 +52,7 @@ import (
 	workloadrayjob "sigs.k8s.io/kueue/pkg/controller/jobs/rayjob"
 	utilpod "sigs.k8s.io/kueue/pkg/util/pod"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+	testingaw "sigs.k8s.io/kueue/pkg/util/testingjobs/appwrapper"
 	testingdeployment "sigs.k8s.io/kueue/pkg/util/testingjobs/deployment"
 	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
 	testingjobset "sigs.k8s.io/kueue/pkg/util/testingjobs/jobset"
@@ -564,6 +567,67 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 			})
 		})
 
+		ginkgo.It("Should run an appwrapper containing a job on worker if admitted", func() {
+			// Since it requires 2 CPU in total, this appwrapper can only be admitted in worker 1.
+			aw := testingaw.MakeAppWrapper("aw", managerNs.Name).
+				Queue(managerLq.Name).
+				Component(testingjob.MakeJob("job-1", managerNs.Name).
+					SetTypeMeta().
+					Suspend(false).
+					Image(util.E2eTestSleepImage, []string{"5s"}). // Give it the time to be observed Active in the live status update step.
+					Parallelism(2).
+					Request(corev1.ResourceCPU, "1").
+					SetTypeMeta().Obj()).
+				Obj()
+
+			ginkgo.By("Creating the appwrapper", func() {
+				gomega.Expect(k8sManagerClient.Create(ctx, aw)).Should(gomega.Succeed())
+			})
+
+			createdWorkload := &kueue.Workload{}
+			wlLookupKey := types.NamespacedName{Name: workloadaw.GetWorkloadNameForAppWrapper(aw.Name, aw.UID), Namespace: managerNs.Name}
+
+			// the execution should be given to worker 1
+			waitForJobAdmitted(wlLookupKey, multiKueueAc.Name, "worker1")
+
+			ginkgo.By("Waiting for the appwrapper to get status updates", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdAppWrapper := &awv1beta2.AppWrapper{}
+					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(aw), createdAppWrapper)).To(gomega.Succeed())
+					g.Expect(createdAppWrapper.Status.Phase).To(gomega.Equal(awv1beta2.AppWrapperRunning))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Waiting for the appwrapper to finish", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sManagerClient.Get(ctx, wlLookupKey, createdWorkload)).To(gomega.Succeed())
+
+					g.Expect(apimeta.FindStatusCondition(createdWorkload.Status.Conditions, kueue.WorkloadFinished)).To(gomega.BeComparableTo(&metav1.Condition{
+						Type:    kueue.WorkloadFinished,
+						Status:  metav1.ConditionTrue,
+						Reason:  kueue.WorkloadFinishedReasonSucceeded,
+						Message: "AppWrapper finished successfully",
+					}, util.IgnoreConditionTimestampsAndObservedGeneration))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Checking no objects are left in the worker clusters and the appwrapper is completed", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					workerWl := &kueue.Workload{}
+					g.Expect(k8sWorker1Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
+					g.Expect(k8sWorker2Client.Get(ctx, wlLookupKey, workerWl)).To(utiltesting.BeNotFoundError())
+					workerAppWrapper := &awv1beta2.AppWrapper{}
+					g.Expect(k8sWorker1Client.Get(ctx, client.ObjectKeyFromObject(aw), workerAppWrapper)).To(utiltesting.BeNotFoundError())
+					g.Expect(k8sWorker2Client.Get(ctx, client.ObjectKeyFromObject(aw), workerAppWrapper)).To(utiltesting.BeNotFoundError())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+				createdAppWrapper := &awv1beta2.AppWrapper{}
+				gomega.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(aw), createdAppWrapper)).To(gomega.Succeed())
+				gomega.Expect(createdAppWrapper.Spec.Suspend).To(gomega.BeFalse())
+				gomega.Expect(createdAppWrapper.Status.Phase).To(gomega.Equal(awv1beta2.AppWrapperSucceeded))
+			})
+		})
+
 		ginkgo.It("Should run a kubeflow PyTorchJob on worker if admitted", func() {
 			// Since it requires 1600M of memory, this job can only be admitted in worker 2.
 			pyTorchJob := testingpytorchjob.MakePyTorchJob("pytorchjob1", managerNs.Name).
diff --git a/test/e2e/multikueue/suite_test.go b/test/e2e/multikueue/suite_test.go
index eb15b1ac3..a7bfe19db 100644
--- a/test/e2e/multikueue/suite_test.go
+++ b/test/e2e/multikueue/suite_test.go
@@ -27,6 +27,7 @@ import (
 	kftraining "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1"
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
+	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	authenticationv1 "k8s.io/api/authentication/v1"
 	batchv1 "k8s.io/api/batch/v1"
@@ -97,6 +98,8 @@ func kubeconfigForMultiKueueSA(ctx context.Context, c client.Client, restConfig
 			policyRule(kftraining.SchemeGroupVersion.Group, "pytorchjobs/status", "get"),
 			policyRule(kftraining.SchemeGroupVersion.Group, "xgboostjobs", resourceVerbs...),
 			policyRule(kftraining.SchemeGroupVersion.Group, "xgboostjobs/status", "get"),
+			policyRule(awv1beta2.GroupVersion.Group, "appwrappers", resourceVerbs...),
+			policyRule(awv1beta2.GroupVersion.Group, "appwrappers/status", "get"),
 			policyRule(kfmpi.SchemeGroupVersion.Group, "mpijobs", resourceVerbs...),
 			policyRule(kfmpi.SchemeGroupVersion.Group, "mpijobs/status", "get"),
 			policyRule(rayv1.SchemeGroupVersion.Group, "rayjobs", resourceVerbs...),
@@ -280,11 +283,14 @@ var _ = ginkgo.BeforeSuite(func() {
 	util.WaitForKubeFlowMPIOperatorAvailability(ctx, k8sWorker1Client)
 	util.WaitForKubeFlowMPIOperatorAvailability(ctx, k8sWorker2Client)
 
+	util.WaitForAppWrapperAvailability(ctx, k8sManagerClient)
+	util.WaitForAppWrapperAvailability(ctx, k8sWorker1Client)
+	util.WaitForAppWrapperAvailability(ctx, k8sWorker2Client)
+
 	util.WaitForKubeRayOperatorAvailability(ctx, k8sWorker1Client)
 	util.WaitForKubeRayOperatorAvailability(ctx, k8sWorker2Client)
 
 	ginkgo.GinkgoLogr.Info("Kueue and all integration operators are available in all the clusters", "waitingTime", time.Since(waitForAvailableStart))
-
 	discoveryClient, err := discovery.NewDiscoveryClientForConfig(managerCfg)
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 	managerK8SVersion, err = kubeversion.FetchServerVersion(discoveryClient)
diff --git a/test/e2e/singlecluster/leaderworkerset_test.go b/test/e2e/singlecluster/leaderworkerset_test.go
index 005bb391b..d4d4ff935 100644
--- a/test/e2e/singlecluster/leaderworkerset_test.go
+++ b/test/e2e/singlecluster/leaderworkerset_test.go
@@ -256,345 +256,280 @@ var _ = ginkgo.Describe("LeaderWorkerSet integration", func() {
 			})
 		})
 
-		ginkgo.It("should allow to scale up", func() {
-			lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
-				Image(util.E2eTestSleepImage, []string{"10m"}).
-				Size(3).
-				Replicas(1).
-				Request(corev1.ResourceCPU, "100m").
-				Queue(lq.Name).
-				Obj()
-
-			ginkgo.By("Create a LeaderWorkerSet", func() {
-				gomega.Expect(k8sClient.Create(ctx, lws)).To(gomega.Succeed())
-			})
-
-			createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
-
-			ginkgo.By("Waiting for replicas is ready", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
-					g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(1)))
-					g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
-						gomega.BeComparableTo(metav1.Condition{
-							Type:    "Available",
-							Status:  metav1.ConditionTrue,
-							Reason:  "AllGroupsReady",
-							Message: "All replicas are ready",
-						}, util.IgnoreConditionTimestampsAndObservedGeneration)),
-					)
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			createdWorkload1 := &kueue.Workload{}
-			wlLookupKey1 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"), Namespace: ns.Name}
-			ginkgo.By("Check workload is created", func() {
-				gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
-			})
-
-			ginkgo.By("Scale up LeaderWorkerSet", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
-					createdLeaderWorkerSet.Spec.Replicas = ptr.To[int32](2)
-					g.Expect(k8sClient.Update(ctx, createdLeaderWorkerSet)).To(gomega.Succeed())
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			ginkgo.By("Waiting for replicas is ready", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
-					g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(2)))
-					g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
-						gomega.BeComparableTo(metav1.Condition{
-							Type:    "Available",
-							Status:  metav1.ConditionTrue,
-							Reason:  "AllGroupsReady",
-							Message: "All replicas are ready",
-						}, util.IgnoreConditionTimestampsAndObservedGeneration)),
-					)
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			ginkgo.By("Check workload for group 1 is still exist", func() {
-				gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
-			})
-
-			createdWorkload2 := &kueue.Workload{}
-			wlLookupKey2 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "1"), Namespace: ns.Name}
-			ginkgo.By("Check workload for group 2 is created", func() {
-				gomega.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).To(gomega.Succeed())
-			})
-
-			ginkgo.By("Delete the LeaderWorkerSet", func() {
-				util.ExpectObjectToBeDeleted(ctx, k8sClient, lws, true)
-			})
-
-			ginkgo.By("Check pods are deleted", func() {
-				pods := &corev1.PodList{}
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
-						leaderworkersetv1.SetNameLabelKey: lws.Name,
-					}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
-					g.Expect(pods.Items).To(gomega.BeEmpty())
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			ginkgo.By("Check workloads are deleted", func() {
-				util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload1, false, util.LongTimeout)
-				util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload2, false, util.LongTimeout)
-			})
-		})
-
-		ginkgo.It("should allow to scale down", func() {
-			lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
-				Image(util.E2eTestSleepImage, []string{"10m"}).
-				Size(3).
-				Replicas(2).
-				Request(corev1.ResourceCPU, "100m").
-				Queue(lq.Name).
-				Obj()
-
-			ginkgo.By("Create a LeaderWorkerSet", func() {
-				gomega.Expect(k8sClient.Create(ctx, lws)).To(gomega.Succeed())
-			})
-
-			createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
-
-			ginkgo.By("Waiting for replicas is ready", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
-					g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(2)))
-					g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
-						gomega.BeComparableTo(metav1.Condition{
-							Type:    "Available",
-							Status:  metav1.ConditionTrue,
-							Reason:  "AllGroupsReady",
-							Message: "All replicas are ready",
-						}, util.IgnoreConditionTimestampsAndObservedGeneration)),
-					)
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			createdWorkload1 := &kueue.Workload{}
-			wlLookupKey1 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"), Namespace: ns.Name}
-			ginkgo.By("Check workload for group 1 is created", func() {
-				gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
-			})
+		ginkgo.DescribeTable("should allow to scale up",
+			func(startupPolicyType leaderworkersetv1.StartupPolicyType) {
+				lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
+					Image(util.E2eTestSleepImage, []string{"10m"}).
+					Size(3).
+					Replicas(1).
+					Request(corev1.ResourceCPU, "100m").
+					Queue(lq.Name).
+					StartupPolicy(startupPolicyType).
+					Obj()
+
+				ginkgo.By("Create a LeaderWorkerSet", func() {
+					gomega.Expect(k8sClient.Create(ctx, lws)).To(gomega.Succeed())
+				})
 
-			createdWorkload2 := &kueue.Workload{}
-			wlLookupKey2 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "1"), Namespace: ns.Name}
-			ginkgo.By("Check workload for group 2 is created", func() {
-				gomega.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).To(gomega.Succeed())
-			})
-
-			ginkgo.By("Scale down LeaderWorkerSet", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
-					createdLeaderWorkerSet.Spec.Replicas = ptr.To[int32](1)
-					g.Expect(k8sClient.Update(ctx, createdLeaderWorkerSet)).To(gomega.Succeed())
-				}, util.Timeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			ginkgo.By("Waiting for replicas is ready", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
-					g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(1)))
-					g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
-						gomega.BeComparableTo(metav1.Condition{
-							Type:    "Available",
-							Status:  metav1.ConditionTrue,
-							Reason:  "AllGroupsReady",
-							Message: "All replicas are ready",
-						}, util.IgnoreConditionTimestampsAndObservedGeneration)),
-					)
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			ginkgo.By("Check workload for group 1 is still exist", func() {
-				gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
-			})
-
-			ginkgo.By("Check workload for group 2 is deleted", func() {
-				util.ExpectObjectToBeDeleted(ctx, k8sClient, createdWorkload2, false)
-			})
-
-			ginkgo.By("Delete the LeaderWorkerSet", func() {
-				util.ExpectObjectToBeDeleted(ctx, k8sClient, lws, true)
-			})
-
-			ginkgo.By("Check pods are deleted", func() {
-				pods := &corev1.PodList{}
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
-						leaderworkersetv1.SetNameLabelKey: lws.Name,
-					}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
-					g.Expect(pods.Items).To(gomega.BeEmpty())
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			ginkgo.By("Check workloads are deleted", func() {
-				util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload1, false, util.LongTimeout)
-			})
-		})
-
-		ginkgo.It("should admit group with multiple leaders and workers that fits and have different resource needs (LeaderCreatedStartupPolicy)", func() {
-			lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
-				Image(util.E2eTestSleepImage, []string{"10m"}).
-				Size(3).
-				Replicas(2).
-				Request(corev1.ResourceCPU, "100m").
-				Queue(lq.Name).
-				LeaderTemplate(corev1.PodTemplateSpec{
-					Spec: corev1.PodSpec{
-						Containers: []corev1.Container{
-							{
-								Name:  "c",
-								Args:  []string{"10m"},
-								Image: util.E2eTestSleepImage,
-								Resources: corev1.ResourceRequirements{
-									Requests: corev1.ResourceList{
-										corev1.ResourceCPU: resource.MustParse("150m"),
-									},
-								},
-							},
-						},
-						NodeSelector: map[string]string{},
-					},
-				}).
-				Obj()
-
-			ginkgo.By("Create a LeaderWorkerSet", func() {
-				gomega.Expect(k8sClient.Create(ctx, lws)).To(gomega.Succeed())
-			})
-
-			createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
-
-			ginkgo.By("Waiting for replicas is ready", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
-					g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(2)))
-					g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
-						gomega.BeComparableTo(metav1.Condition{
-							Type:    "Available",
-							Status:  metav1.ConditionTrue,
-							Reason:  "AllGroupsReady",
-							Message: "All replicas are ready",
-						}, util.IgnoreConditionTimestampsAndObservedGeneration)),
-					)
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			createdWorkload1 := &kueue.Workload{}
-			wlLookupKey1 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"), Namespace: ns.Name}
-			ginkgo.By("Check workload for group 1 is created", func() {
-				gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
-			})
-
-			createdWorkload2 := &kueue.Workload{}
-			wlLookupKey2 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"), Namespace: ns.Name}
-			ginkgo.By("Check workload for group 2 is created", func() {
-				gomega.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).To(gomega.Succeed())
-			})
-
-			ginkgo.By("Delete the LeaderWorkerSet", func() {
-				util.ExpectObjectToBeDeleted(ctx, k8sClient, lws, true)
-			})
+				createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
 
-			ginkgo.By("Check pods are deleted", func() {
-				pods := &corev1.PodList{}
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
-						leaderworkersetv1.SetNameLabelKey: lws.Name,
-					}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
-					g.Expect(pods.Items).To(gomega.BeEmpty())
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
+				ginkgo.By("Waiting for replicas is ready", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+						g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(1)))
+						g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
+							gomega.BeComparableTo(metav1.Condition{
+								Type:    "Available",
+								Status:  metav1.ConditionTrue,
+								Reason:  "AllGroupsReady",
+								Message: "All replicas are ready",
+							}, util.IgnoreConditionTimestampsAndObservedGeneration)),
+						)
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				createdWorkload1 := &kueue.Workload{}
+				wlLookupKey1 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"), Namespace: ns.Name}
+				ginkgo.By("Check workload is created", func() {
+					gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
+				})
+
+				ginkgo.By("Scale up LeaderWorkerSet", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+						createdLeaderWorkerSet.Spec.Replicas = ptr.To[int32](2)
+						g.Expect(k8sClient.Update(ctx, createdLeaderWorkerSet)).To(gomega.Succeed())
+					}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Waiting for replicas is ready", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+						g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(2)))
+						g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
+							gomega.BeComparableTo(metav1.Condition{
+								Type:    "Available",
+								Status:  metav1.ConditionTrue,
+								Reason:  "AllGroupsReady",
+								Message: "All replicas are ready",
+							}, util.IgnoreConditionTimestampsAndObservedGeneration)),
+						)
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Check workload for group 1 is still exist", func() {
+					gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
+				})
+
+				createdWorkload2 := &kueue.Workload{}
+				wlLookupKey2 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "1"), Namespace: ns.Name}
+				ginkgo.By("Check workload for group 2 is created", func() {
+					gomega.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).To(gomega.Succeed())
+				})
+
+				ginkgo.By("Delete the LeaderWorkerSet", func() {
+					util.ExpectObjectToBeDeleted(ctx, k8sClient, lws, true)
+				})
+
+				ginkgo.By("Check pods are deleted", func() {
+					pods := &corev1.PodList{}
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
+							leaderworkersetv1.SetNameLabelKey: lws.Name,
+						}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
+						g.Expect(pods.Items).To(gomega.BeEmpty())
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Check workloads are deleted", func() {
+					util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload1, false, util.LongTimeout)
+					util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload2, false, util.LongTimeout)
+				})
+			},
+			ginkgo.Entry("LeaderCreatedStartupPolicy", leaderworkersetv1.LeaderCreatedStartupPolicy),
+			ginkgo.Entry("LeaderReadyStartupPolicy", leaderworkersetv1.LeaderReadyStartupPolicy),
+		)
+
+		ginkgo.DescribeTable("should allow to scale down",
+			func(startupPolicyType leaderworkersetv1.StartupPolicyType) {
+				lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
+					Image(util.E2eTestSleepImage, []string{"10m"}).
+					Size(3).
+					Replicas(2).
+					Request(corev1.ResourceCPU, "100m").
+					Queue(lq.Name).
+					StartupPolicy(startupPolicyType).
+					Obj()
+
+				ginkgo.By("Create a LeaderWorkerSet", func() {
+					gomega.Expect(k8sClient.Create(ctx, lws)).To(gomega.Succeed())
+				})
 
-			ginkgo.By("Check workloads are deleted", func() {
-				util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload1, false, util.LongTimeout)
-				util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload2, false, util.LongTimeout)
-			})
-		})
+				createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
 
-		ginkgo.It("should admit group with multiple leaders and workers that fits and have different resource needs (LeaderReadyStartupPolicy)", func() {
-			lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
-				Image(util.E2eTestSleepImage, []string{"10m"}).
-				Size(3).
-				Replicas(2).
-				Request(corev1.ResourceCPU, "100m").
-				StartupPolicy(leaderworkersetv1.LeaderReadyStartupPolicy).
-				Queue(lq.Name).
-				LeaderTemplate(corev1.PodTemplateSpec{
-					Spec: corev1.PodSpec{
-						Containers: []corev1.Container{
-							{
-								Name:  "c",
-								Args:  []string{"10m"},
-								Image: util.E2eTestSleepImage,
-								Resources: corev1.ResourceRequirements{
-									Requests: corev1.ResourceList{
-										corev1.ResourceCPU: resource.MustParse("150m"),
+				ginkgo.By("Waiting for replicas is ready", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+						g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(2)))
+						g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
+							gomega.BeComparableTo(metav1.Condition{
+								Type:    "Available",
+								Status:  metav1.ConditionTrue,
+								Reason:  "AllGroupsReady",
+								Message: "All replicas are ready",
+							}, util.IgnoreConditionTimestampsAndObservedGeneration)),
+						)
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				createdWorkload1 := &kueue.Workload{}
+				wlLookupKey1 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"), Namespace: ns.Name}
+				ginkgo.By("Check workload for group 1 is created", func() {
+					gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
+				})
+
+				createdWorkload2 := &kueue.Workload{}
+				wlLookupKey2 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "1"), Namespace: ns.Name}
+				ginkgo.By("Check workload for group 2 is created", func() {
+					gomega.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).To(gomega.Succeed())
+				})
+
+				ginkgo.By("Scale down LeaderWorkerSet", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+						createdLeaderWorkerSet.Spec.Replicas = ptr.To[int32](1)
+						g.Expect(k8sClient.Update(ctx, createdLeaderWorkerSet)).To(gomega.Succeed())
+					}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Waiting for replicas is ready", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+						g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(1)))
+						g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
+							gomega.BeComparableTo(metav1.Condition{
+								Type:    "Available",
+								Status:  metav1.ConditionTrue,
+								Reason:  "AllGroupsReady",
+								Message: "All replicas are ready",
+							}, util.IgnoreConditionTimestampsAndObservedGeneration)),
+						)
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Check workload for group 1 is still exist", func() {
+					gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
+				})
+
+				ginkgo.By("Check workload for group 2 is deleted", func() {
+					util.ExpectObjectToBeDeleted(ctx, k8sClient, createdWorkload2, false)
+				})
+
+				ginkgo.By("Delete the LeaderWorkerSet", func() {
+					util.ExpectObjectToBeDeleted(ctx, k8sClient, lws, true)
+				})
+
+				ginkgo.By("Check pods are deleted", func() {
+					pods := &corev1.PodList{}
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
+							leaderworkersetv1.SetNameLabelKey: lws.Name,
+						}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
+						g.Expect(pods.Items).To(gomega.BeEmpty())
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Check workloads are deleted", func() {
+					util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload1, false, util.LongTimeout)
+				})
+			},
+			ginkgo.Entry("LeaderCreatedStartupPolicy", leaderworkersetv1.LeaderCreatedStartupPolicy),
+			ginkgo.Entry("LeaderReadyStartupPolicy", leaderworkersetv1.LeaderReadyStartupPolicy),
+		)
+
+		ginkgo.DescribeTable("should admit group with multiple leaders and workers that fits and have different resource needs",
+			func(startupPolicyType leaderworkersetv1.StartupPolicyType) {
+				lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
+					Image(util.E2eTestSleepImage, []string{"10m"}).
+					Size(3).
+					Replicas(2).
+					Request(corev1.ResourceCPU, "100m").
+					Queue(lq.Name).
+					LeaderTemplate(corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name:  "c",
+									Args:  []string{"10m"},
+									Image: util.E2eTestSleepImage,
+									Resources: corev1.ResourceRequirements{
+										Requests: corev1.ResourceList{
+											corev1.ResourceCPU: resource.MustParse("150m"),
+										},
 									},
 								},
 							},
+							NodeSelector: map[string]string{},
 						},
-						NodeSelector: map[string]string{},
-					},
-				}).
-				Obj()
-
-			ginkgo.By("Create a LeaderWorkerSet", func() {
-				gomega.Expect(k8sClient.Create(ctx, lws)).To(gomega.Succeed())
-			})
-
-			createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
-
-			ginkgo.By("Waiting for replicas is ready", func() {
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
-					g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(2)))
-					g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
-						gomega.BeComparableTo(metav1.Condition{
-							Type:    "Available",
-							Status:  metav1.ConditionTrue,
-							Reason:  "AllGroupsReady",
-							Message: "All replicas are ready",
-						}, util.IgnoreConditionTimestampsAndObservedGeneration)),
-					)
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
-
-			createdWorkload1 := &kueue.Workload{}
-			wlLookupKey1 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"), Namespace: ns.Name}
-			ginkgo.By("Check workload for group 1 is created", func() {
-				gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
-			})
+					}).
+					StartupPolicy(startupPolicyType).
+					Obj()
 
-			createdWorkload2 := &kueue.Workload{}
-			wlLookupKey2 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"), Namespace: ns.Name}
-			ginkgo.By("Check workload for group 2 is created", func() {
-				gomega.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).To(gomega.Succeed())
-			})
+				ginkgo.By("Create a LeaderWorkerSet", func() {
+					gomega.Expect(k8sClient.Create(ctx, lws)).To(gomega.Succeed())
+				})
 
-			ginkgo.By("Delete the LeaderWorkerSet", func() {
-				util.ExpectObjectToBeDeleted(ctx, k8sClient, lws, true)
-			})
-
-			ginkgo.By("Check pods are deleted", func() {
-				pods := &corev1.PodList{}
-				gomega.Eventually(func(g gomega.Gomega) {
-					g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
-						leaderworkersetv1.SetNameLabelKey: lws.Name,
-					}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
-					g.Expect(pods.Items).To(gomega.BeEmpty())
-				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
-			})
+				createdLeaderWorkerSet := &leaderworkersetv1.LeaderWorkerSet{}
 
-			ginkgo.By("Check workloads are deleted", func() {
-				util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload1, false, util.LongTimeout)
-				util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload2, false, util.LongTimeout)
-			})
-		})
+				ginkgo.By("Waiting for replicas is ready", func() {
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lws), createdLeaderWorkerSet)).To(gomega.Succeed())
+						g.Expect(createdLeaderWorkerSet.Status.ReadyReplicas).To(gomega.Equal(int32(2)))
+						g.Expect(createdLeaderWorkerSet.Status.Conditions).To(gomega.ContainElement(
+							gomega.BeComparableTo(metav1.Condition{
+								Type:    "Available",
+								Status:  metav1.ConditionTrue,
+								Reason:  "AllGroupsReady",
+								Message: "All replicas are ready",
+							}, util.IgnoreConditionTimestampsAndObservedGeneration)),
+						)
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				createdWorkload1 := &kueue.Workload{}
+				wlLookupKey1 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"), Namespace: ns.Name}
+				ginkgo.By("Check workload for group 1 is created", func() {
+					gomega.Expect(k8sClient.Get(ctx, wlLookupKey1, createdWorkload1)).To(gomega.Succeed())
+				})
+
+				createdWorkload2 := &kueue.Workload{}
+				wlLookupKey2 := types.NamespacedName{Name: leaderworkerset.GetWorkloadName(lws.UID, lws.Name, "0"), Namespace: ns.Name}
+				ginkgo.By("Check workload for group 2 is created", func() {
+					gomega.Expect(k8sClient.Get(ctx, wlLookupKey2, createdWorkload2)).To(gomega.Succeed())
+				})
+
+				ginkgo.By("Delete the LeaderWorkerSet", func() {
+					util.ExpectObjectToBeDeleted(ctx, k8sClient, lws, true)
+				})
+
+				ginkgo.By("Check pods are deleted", func() {
+					pods := &corev1.PodList{}
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sClient.List(ctx, pods, client.MatchingLabels{
+							leaderworkersetv1.SetNameLabelKey: lws.Name,
+						}, client.InNamespace(lws.Namespace))).Should(gomega.Succeed())
+						g.Expect(pods.Items).To(gomega.BeEmpty())
+					}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("Check workloads are deleted", func() {
+					util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload1, false, util.LongTimeout)
+					util.ExpectObjectToBeDeletedWithTimeout(ctx, k8sClient, createdWorkload2, false, util.LongTimeout)
+				})
+			},
+			ginkgo.Entry("LeaderCreatedStartupPolicy", leaderworkersetv1.LeaderCreatedStartupPolicy),
+			ginkgo.Entry("LeaderReadyStartupPolicy", leaderworkersetv1.LeaderReadyStartupPolicy),
+		)
 
 		ginkgo.It("should allow to update the PodTemplate in LeaderWorkerSet", func() {
 			lws := leaderworkersettesting.MakeLeaderWorkerSet("lws", ns.Name).
diff --git a/test/e2e/tas/appwrapper_test.go b/test/e2e/tas/appwrapper_test.go
new file mode 100644
index 000000000..cf399d0ba
--- /dev/null
+++ b/test/e2e/tas/appwrapper_test.go
@@ -0,0 +1,132 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tase2e
+
+import (
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/util/testing"
+	awtesting "sigs.k8s.io/kueue/pkg/util/testingjobs/appwrapper"
+	utiltestingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
+	"sigs.k8s.io/kueue/test/util"
+)
+
+var _ = ginkgo.Describe("TopologyAwareScheduling for AppWrapper", func() {
+	var (
+		ns           *corev1.Namespace
+		topology     *kueuealpha.Topology
+		tasFlavor    *kueue.ResourceFlavor
+		clusterQueue *kueue.ClusterQueue
+		localQueue   *kueue.LocalQueue
+	)
+
+	ginkgo.BeforeEach(func() {
+		ns = &corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				GenerateName: "e2e-tas-aw-",
+			},
+		}
+		gomega.Expect(k8sClient.Create(ctx, ns)).To(gomega.Succeed())
+
+		topology = testing.MakeDefaultThreeLevelTopology("datacenter")
+		gomega.Expect(k8sClient.Create(ctx, topology)).Should(gomega.Succeed())
+
+		tasFlavor = testing.MakeResourceFlavor("tas-flavor").
+			NodeLabel(tasNodeGroupLabel, instanceType).
+			TopologyName(topology.Name).
+			Obj()
+		gomega.Expect(k8sClient.Create(ctx, tasFlavor)).Should(gomega.Succeed())
+
+		clusterQueue = testing.MakeClusterQueue("cluster-queue").
+			ResourceGroup(
+				*testing.MakeFlavorQuotas(tasFlavor.Name).
+					Resource(corev1.ResourceCPU, "1").
+					Resource(extraResource, "8").
+					Obj(),
+			).
+			Obj()
+		gomega.Expect(k8sClient.Create(ctx, clusterQueue)).Should(gomega.Succeed())
+		util.ExpectClusterQueuesToBeActive(ctx, k8sClient, clusterQueue)
+
+		localQueue = testing.MakeLocalQueue("local-queue", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+		gomega.Expect(k8sClient.Create(ctx, localQueue)).Should(gomega.Succeed())
+		util.ExpectLocalQueuesToBeActive(ctx, k8sClient, localQueue)
+	})
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteAllAppWrappersInNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+		gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, tasFlavor, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, topology, true)
+		util.ExpectAllPodsInNamespaceDeleted(ctx, k8sClient, ns)
+	})
+
+	ginkgo.When("Creating an AppWrapper", func() {
+		ginkgo.It("Should place pods", func() {
+			numPods := 4
+
+			aw := awtesting.MakeAppWrapper("appwrapper", ns.Name).
+				Component(utiltestingjob.MakeJob("job-0", ns.Name).
+					Parallelism(int32(numPods)).
+					Completions(int32(numPods)).
+					Request(corev1.ResourceCPU, "100m").
+					Limit(corev1.ResourceCPU, "100m").
+					Request(extraResource, "1").
+					Limit(extraResource, "1").
+					Suspend(false).
+					Image(util.E2eTestSleepImage, []string{"60s"}).
+					PodAnnotation(kueuealpha.PodSetPreferredTopologyAnnotation, testing.DefaultRackTopologyLevel).
+					SetTypeMeta().Obj()).
+				Queue(localQueue.Name).
+				Obj()
+
+			gomega.Expect(k8sClient.Create(ctx, aw)).Should(gomega.Succeed())
+
+			ginkgo.By("AppWrapper is unsuspended", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(aw), aw)).To(gomega.Succeed())
+					g.Expect(aw.Spec.Suspend).Should(gomega.BeFalse())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			pods := &corev1.PodList{}
+			ginkgo.By("ensure all pods are created", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.List(ctx, pods, client.InNamespace(ns.Name))).To(gomega.Succeed())
+					g.Expect(pods.Items).Should(gomega.HaveLen(numPods))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("ensure all pods are scheduled", func() {
+				listOpts := &client.ListOptions{
+					FieldSelector: fields.OneTermNotEqualSelector("spec.nodeName", ""),
+				}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.List(ctx, pods, client.InNamespace(ns.Name), listOpts)).To(gomega.Succeed())
+					g.Expect(pods.Items).Should(gomega.HaveLen(numPods))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+	})
+})
diff --git a/test/e2e/tas/suite_test.go b/test/e2e/tas/suite_test.go
index 885e7a2a7..34168f675 100644
--- a/test/e2e/tas/suite_test.go
+++ b/test/e2e/tas/suite_test.go
@@ -61,6 +61,7 @@ var _ = ginkgo.BeforeSuite(func() {
 	util.WaitForJobSetAvailability(ctx, k8sClient)
 	util.WaitForKubeFlowTrainingOperatorAvailability(ctx, k8sClient)
 	util.WaitForKubeFlowMPIOperatorAvailability(ctx, k8sClient)
+	util.WaitForAppWrapperAvailability(ctx, k8sClient)
 	ginkgo.GinkgoLogr.Info(
 		"Kueue, JobSet, KubeFlow Training and KubeFlow MPI operators are available in the cluster",
 		"waitingTime", time.Since(waitForAvailableStart),
diff --git a/test/integration/multikueue/multikueue_test.go b/test/integration/multikueue/multikueue_test.go
index 920dac9d5..d735389f9 100644
--- a/test/integration/multikueue/multikueue_test.go
+++ b/test/integration/multikueue/multikueue_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
 	gomegatypes "github.com/onsi/gomega/types"
+	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -46,6 +47,7 @@ import (
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/admissionchecks/multikueue"
+	workloadappwrapper "sigs.k8s.io/kueue/pkg/controller/jobs/appwrapper"
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
 	workloadjobset "sigs.k8s.io/kueue/pkg/controller/jobs/jobset"
 	workloadpaddlejob "sigs.k8s.io/kueue/pkg/controller/jobs/kubeflow/jobs/paddlejob"
@@ -58,6 +60,7 @@ import (
 	workloadrayjob "sigs.k8s.io/kueue/pkg/controller/jobs/rayjob"
 	"sigs.k8s.io/kueue/pkg/features"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+	testingaw "sigs.k8s.io/kueue/pkg/util/testingjobs/appwrapper"
 	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
 	testingjobset "sigs.k8s.io/kueue/pkg/util/testingjobs/jobset"
 	testingmpijob "sigs.k8s.io/kueue/pkg/util/testingjobs/mpijob"
@@ -77,7 +80,7 @@ var defaultEnabledIntegrations sets.Set[string] = sets.New(
 	"batch/job", "kubeflow.org/mpijob", "ray.io/rayjob", "ray.io/raycluster",
 	"jobset.x-k8s.io/jobset", "kubeflow.org/paddlejob",
 	"kubeflow.org/pytorchjob", "kubeflow.org/tfjob", "kubeflow.org/xgboostjob",
-	"pod")
+	"pod", "workload.codeflare.dev/appwrapper")
 
 var _ = ginkgo.Describe("Multikueue", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
 	var (
@@ -1107,6 +1110,51 @@ var _ = ginkgo.Describe("Multikueue", ginkgo.Ordered, ginkgo.ContinueOnFailure,
 		})
 	})
 
+	ginkgo.It("Should run an appwrapper on worker if admitted", func() {
+		aw := testingaw.MakeAppWrapper("aw", managerNs.Name).
+			Component(testingjob.MakeJob("job-1", managerNs.Name).SetTypeMeta().Parallelism(1).Obj()).
+			Queue(managerLq.Name).
+			ManagedBy(kueue.MultiKueueControllerName).
+			Obj()
+
+		gomega.Expect(managerTestCluster.client.Create(managerTestCluster.ctx, aw)).Should(gomega.Succeed())
+		wlLookupKey := types.NamespacedName{Name: workloadappwrapper.GetWorkloadNameForAppWrapper(aw.Name, aw.UID), Namespace: managerNs.Name}
+
+		admission := utiltesting.MakeAdmission(managerCq.Name).PodSets(
+			kueue.PodSetAssignment{
+				Name: "aw-0-0",
+			},
+		)
+
+		admitWorkloadAndCheckWorkerCopies(multikueueAC.Name, wlLookupKey, admission)
+
+		ginkgo.By("changing the status of the appwrapper in the worker, updates the manager's appwrappers status", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				createdAppWrapper := awv1beta2.AppWrapper{}
+				g.Expect(worker2TestCluster.client.Get(worker2TestCluster.ctx, client.ObjectKeyFromObject(aw), &createdAppWrapper)).To(gomega.Succeed())
+				createdAppWrapper.Status.Phase = awv1beta2.AppWrapperRunning
+				g.Expect(worker2TestCluster.client.Status().Update(worker2TestCluster.ctx, &createdAppWrapper)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			gomega.Eventually(func(g gomega.Gomega) {
+				createdAppWrapper := awv1beta2.AppWrapper{}
+				g.Expect(managerTestCluster.client.Get(managerTestCluster.ctx, client.ObjectKeyFromObject(aw), &createdAppWrapper)).To(gomega.Succeed())
+				g.Expect(createdAppWrapper.Status.Phase).To(gomega.Equal(awv1beta2.AppWrapperRunning))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("finishing the worker appwrapper, the manager's wl is marked as finished and the worker2 wl removed", func() {
+			finishJobReason := "AppWrapper finished successfully"
+			gomega.Eventually(func(g gomega.Gomega) {
+				createdAppWrapper := awv1beta2.AppWrapper{}
+				g.Expect(worker2TestCluster.client.Get(worker2TestCluster.ctx, client.ObjectKeyFromObject(aw), &createdAppWrapper)).To(gomega.Succeed())
+				createdAppWrapper.Status.Phase = awv1beta2.AppWrapperSucceeded
+				g.Expect(worker2TestCluster.client.Status().Update(worker2TestCluster.ctx, &createdAppWrapper)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			waitForWorkloadToFinishAndRemoteWorkloadToBeDeleted(wlLookupKey, finishJobReason)
+		})
+	})
+
 	ginkgo.It("Should not run a MPIJob on worker if set to be managed by external controller", func() {
 		admission := utiltesting.MakeAdmission(managerCq.Name).PodSets(
 			kueue.PodSetAssignment{
diff --git a/test/integration/multikueue/suite_test.go b/test/integration/multikueue/suite_test.go
index f27ec7be1..1e17a49b2 100644
--- a/test/integration/multikueue/suite_test.go
+++ b/test/integration/multikueue/suite_test.go
@@ -41,6 +41,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/controller/core"
 	"sigs.k8s.io/kueue/pkg/controller/core/indexer"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+	workloadaw "sigs.k8s.io/kueue/pkg/controller/jobs/appwrapper"
 	workloadjob "sigs.k8s.io/kueue/pkg/controller/jobs/job"
 	workloadjobset "sigs.k8s.io/kueue/pkg/controller/jobs/jobset"
 	workloadpaddlejob "sigs.k8s.io/kueue/pkg/controller/jobs/kubeflow/jobs/paddlejob"
@@ -99,6 +100,7 @@ func createCluster(setupFnc framework.ManagerSetup, apiFeatureGates ...string) c
 			util.TrainingOperatorCrds,
 			util.MpiOperatorCrds,
 			util.RayOperatorCrds,
+			util.AppWrapperCrds,
 		},
 		APIServerFeatureGates: apiFeatureGates,
 	}
@@ -258,6 +260,18 @@ func managerSetup(ctx context.Context, mgr manager.Manager) {
 
 	err = workloadraycluster.SetupRayClusterWebhook(mgr, jobframework.WithCache(cCache), jobframework.WithQueues(queues))
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+	err = workloadaw.SetupIndexes(ctx, mgr.GetFieldIndexer())
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+	appwrapperReconciler := workloadaw.NewReconciler(
+		mgr.GetClient(),
+		mgr.GetEventRecorderFor(constants.JobControllerName))
+	err = appwrapperReconciler.SetupWithManager(mgr)
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+
+	err = workloadaw.SetupAppWrapperWebhook(mgr, jobframework.WithCache(cCache), jobframework.WithQueues(queues))
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 }
 
 func managerAndMultiKueueSetup(ctx context.Context, mgr manager.Manager, gcInterval time.Duration, enabledIntegrations sets.Set[string]) {
diff --git a/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go b/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go
index c61c684e1..f9856ed4a 100644
--- a/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/appwrapper/appwrapper_controller_test.go
@@ -20,25 +20,31 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
 	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
 	corev1 "k8s.io/api/core/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/utils/clock"
+	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	configapi "sigs.k8s.io/kueue/apis/config/v1beta1"
+	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 	workloadaw "sigs.k8s.io/kueue/pkg/controller/jobs/appwrapper"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	testingaw "sigs.k8s.io/kueue/pkg/util/testingjobs/appwrapper"
 	utiltestingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
+	testingnode "sigs.k8s.io/kueue/pkg/util/testingjobs/node"
 	"sigs.k8s.io/kueue/pkg/workload"
 	"sigs.k8s.io/kueue/test/util"
 )
@@ -833,3 +839,130 @@ var _ = ginkgo.Describe("AppWrapper controller interacting with scheduler", gink
 		util.ExpectReservingActiveWorkloadsMetric(clusterQueue, 1)
 	})
 })
+
+var _ = ginkgo.Describe("AppWrapper controller when TopologyAwareScheduling enabled", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
+	const (
+		nodeGroupLabel = "node-group"
+		tasBlockLabel  = "cloud.com/topology-block"
+	)
+
+	var (
+		ns           *corev1.Namespace
+		nodes        []corev1.Node
+		topology     *kueuealpha.Topology
+		tasFlavor    *kueue.ResourceFlavor
+		clusterQueue *kueue.ClusterQueue
+		localQueue   *kueue.LocalQueue
+	)
+
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, managerAndSchedulerSetup(true))
+	})
+
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
+
+	ginkgo.BeforeEach(func() {
+		features.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), features.TopologyAwareScheduling, true)
+
+		ns = &corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				GenerateName: "tas-aw-",
+			},
+		}
+		gomega.Expect(k8sClient.Create(ctx, ns)).To(gomega.Succeed())
+
+		nodes = []corev1.Node{
+			*testingnode.MakeNode("b1").
+				Label("node-group", "tas").
+				Label(tasBlockLabel, "b1").
+				StatusAllocatable(corev1.ResourceList{
+					corev1.ResourceCPU:    resource.MustParse("1"),
+					corev1.ResourceMemory: resource.MustParse("1Gi"),
+				}).
+				Ready().
+				Obj(),
+		}
+		util.CreateNodes(ctx, k8sClient, nodes)
+
+		topology = testing.MakeTopology("default").Levels(tasBlockLabel).Obj()
+		gomega.Expect(k8sClient.Create(ctx, topology)).Should(gomega.Succeed())
+
+		tasFlavor = testing.MakeResourceFlavor("tas-flavor").
+			NodeLabel(nodeGroupLabel, "tas").
+			TopologyName("default").Obj()
+		gomega.Expect(k8sClient.Create(ctx, tasFlavor)).Should(gomega.Succeed())
+
+		clusterQueue = testing.MakeClusterQueue("cluster-queue").
+			ResourceGroup(*testing.MakeFlavorQuotas(tasFlavor.Name).Resource(corev1.ResourceCPU, "5").Obj()).
+			Obj()
+		gomega.Expect(k8sClient.Create(ctx, clusterQueue)).Should(gomega.Succeed())
+		util.ExpectClusterQueuesToBeActive(ctx, k8sClient, clusterQueue)
+
+		localQueue = testing.MakeLocalQueue("local-queue", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+		gomega.Expect(k8sClient.Create(ctx, localQueue)).Should(gomega.Succeed())
+	})
+
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, tasFlavor, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, topology, true)
+		for _, node := range nodes {
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, &node, true)
+		}
+	})
+
+	ginkgo.It("should admit workload which fits in a required topology domain", func() {
+		aw := testingaw.MakeAppWrapper(awName, ns.Name).
+			Component(utiltestingjob.MakeJob("job", ns.Name).
+				PodAnnotation(kueuealpha.PodSetRequiredTopologyAnnotation, tasBlockLabel).
+				Request(corev1.ResourceCPU, "1").
+				SetTypeMeta().
+				Obj()).
+			Queue(localQueue.Name).
+			Suspend(false).
+			Obj()
+		ginkgo.By("creating a job which requires block", func() {
+			gomega.Expect(k8sClient.Create(ctx, aw)).Should(gomega.Succeed())
+		})
+
+		wl := &kueue.Workload{}
+		wlLookupKey := types.NamespacedName{Name: workloadaw.GetWorkloadNameForAppWrapper(aw.Name, aw.UID), Namespace: ns.Name}
+
+		ginkgo.By("verify the workload is created", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlLookupKey, wl)).Should(gomega.Succeed())
+				g.Expect(wl.Spec.PodSets).Should(gomega.BeComparableTo([]kueue.PodSet{{
+					Name:  wl.Spec.PodSets[0].Name,
+					Count: 1,
+					TopologyRequest: &kueue.PodSetTopologyRequest{
+						Required: ptr.To(tasBlockLabel),
+					},
+				}}, cmpopts.IgnoreFields(kueue.PodSet{}, "Template")))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("verify the workload is admitted", func() {
+			util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl)
+			util.ExpectReservingActiveWorkloadsMetric(clusterQueue, 1)
+		})
+
+		ginkgo.By("verify admission for the workload", func() {
+			wl := &kueue.Workload{}
+			wlLookupKey := types.NamespacedName{Name: workloadaw.GetWorkloadNameForAppWrapper(aw.Name, aw.UID), Namespace: ns.Name}
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlLookupKey, wl)).Should(gomega.Succeed())
+				g.Expect(wl.Status.Admission).ShouldNot(gomega.BeNil())
+				g.Expect(wl.Status.Admission.PodSetAssignments).Should(gomega.HaveLen(1))
+				g.Expect(wl.Status.Admission.PodSetAssignments[0].TopologyAssignment).Should(gomega.BeComparableTo(
+					&kueue.TopologyAssignment{
+						Levels:  []string{tasBlockLabel},
+						Domains: []kueue.TopologyDomainAssignment{{Count: 1, Values: []string{"b1"}}},
+					},
+				))
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+	})
+})
diff --git a/test/util/util.go b/test/util/util.go
index 8a9ae8757..d5f3c1b7c 100644
--- a/test/util/util.go
+++ b/test/util/util.go
@@ -33,6 +33,7 @@ import (
 	kftraining "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1"
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
+	awv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
 	"github.com/prometheus/client_golang/prometheus"
 	zaplog "go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
@@ -176,6 +177,14 @@ func DeleteAllPyTorchJobsInNamespace(ctx context.Context, c client.Client, ns *c
 	return nil
 }
 
+func DeleteAllAppWrappersInNamespace(ctx context.Context, c client.Client, ns *corev1.Namespace) error {
+	err := c.DeleteAllOf(ctx, &awv1beta2.AppWrapper{}, client.InNamespace(ns.Name), client.PropagationPolicy(metav1.DeletePropagationBackground))
+	if err != nil && !apierrors.IsNotFound(err) && !errors.Is(err, &apimeta.NoKindMatchError{}) {
+		return err
+	}
+	return nil
+}
+
 func DeleteAllPodsInNamespace(ctx context.Context, c client.Client, ns *corev1.Namespace) error {
 	return deleteAllPodsInNamespace(ctx, c, ns, 2)
 }