Repository: kubernetes-sigs/kueue

Issue:
TAS: Two-level JobSet scheduling
<!-- Please only use this template for submitting enhancement requests -->

**What would you like to be added**:
A possibility to define topology for each Job in ReplicatedJob to co-locate the whole ReplicatedJob in one domain and each Job in that ReplicatedJob in lower-level domain within that domain.

In more generic terms - an API to define required sub PodSet topology and allow for two-level scheduling for the whole PodSet and lower-level topology domain for each chunk within that PodSet.

An example is a JobSet in which user can define topology for all Jobs resulting from a single ReplicatedJob. We want to allow user to define topology for each such Job and possibly co-locate all such Jobs in higher-level topology.

**Why is this needed**:
Currently Topology-Aware Scheduling considers the whole PodSet as a single unit for which Kueue looks for a matching domain. For JobSet PodSet covers all pods resulting from the whole ReplicatedJob. However, there are use-cases in which user would like to co-locate each Job in a lower-level domains and co-locate all those Jobs in higher-level domain.

Additionally, user would like to define a topology only for each Job without any expectations for a topology of the whole ReplicatedJob.

**Completion requirements**:

This enhancement requires the following artifacts:

- [ ] Design doc
- [x] API change
- [ ] Docs update

The artifacts should be linked in subsequent comments.


Hints from issue comments:
@mimowo: /retitle TAS: Two-level JobSet scheduling@lchrzaszcz: TAS KEP update to reflect two-level scheduling and podset chunk topology: https://github.com/kubernetes-sigs/kueue/pull/5449@kerthcet: One quick question: can this solve the https://github.com/kubernetes-sigs/kueue/pull/3941? E.g. co-locate the master and worker jobs in pytorchjob.
Thanks!@kerthcet: I'll read the KEP in detail later.@kerthcet: This focused on podSet chunks, not between multi-podsets. @lchrzaszcz: I will be looking into cross-podsets scheduling in few following weeks. I'll do it mostly for LeaderWorkerSet though, but I think the underlying mechanism will be universal, so probably it'll be easy to add pytorchjob support afterwards.@kerthcet: Thanks @lchrzaszcz for your response, I think leaderworkerset pytorchjob are just two different workload types, the mechanism should be universal to them all I believe. Glad to jump into the discussion.@tenzen-y: /assign @lchrzaszcz 

Solution patch:
diff --git a/apis/config/v1beta1/configuration_types.go b/apis/config/v1beta1/configuration_types.go
index 9bcfbf7ff..f8705932a 100644
--- a/apis/config/v1beta1/configuration_types.go
+++ b/apis/config/v1beta1/configuration_types.go
@@ -40,7 +40,7 @@ type Configuration struct {
 	ControllerManager `json:",inline"`
 
 	// ManageJobsWithoutQueueName controls whether or not Kueue reconciles
-	// jobs that don't set the annotation kueue.x-k8s.io/queue-name.
+	// jobs that don't set the label kueue.x-k8s.io/queue-name.
 	// If set to true, then those jobs will be suspended and never started unless
 	// they are assigned a queue and eventually admitted. This also applies to
 	// jobs created before starting the kueue controller.
diff --git a/hack/e2e-common.sh b/hack/e2e-common.sh
index cbe78e32a..05c919056 100644
--- a/hack/e2e-common.sh
+++ b/hack/e2e-common.sh
@@ -68,23 +68,29 @@ export E2E_TEST_AGNHOST_IMAGE=registry.k8s.io/e2e-test-images/agnhost:2.53@sha25
 E2E_TEST_AGNHOST_IMAGE_WITHOUT_SHA=${E2E_TEST_AGNHOST_IMAGE%%@*}
 
 
-# $1 - cluster name
+# $1 cluster name
+# $2 kubeconfig
 function cluster_cleanup {
-	kubectl config use-context "kind-$1"
-        $KIND export logs "$ARTIFACTS" --name "$1" || true
-        kubectl describe pods -n kueue-system > "$ARTIFACTS/$1-kueue-system-pods.log" || true
-        kubectl describe pods > "$ARTIFACTS/$1-default-pods.log" || true
-        $KIND delete cluster --name "$1"
+    kubectl config --kubeconfig="$2" use-context "kind-$1"
+    
+    $KIND export logs "$ARTIFACTS" --name "$1" || true
+    kubectl describe pods --kubeconfig="$2" -n kueue-system > "$ARTIFACTS/$1-kueue-system-pods.log" || true
+    kubectl describe pods --kubeconfig="$2" > "$ARTIFACTS/$1-default-pods.log" || true
+    $KIND delete cluster --name "$1"
 }
 
 # $1 cluster name
 # $2 cluster kind config
+# $3 kubeconfig
 function cluster_create {
-        $KIND create cluster --name "$1" --image "$E2E_KIND_VERSION" --config "$2" --wait 1m -v 5  > "$ARTIFACTS/$1-create.log" 2>&1 \
-		||  { echo "unable to start the $1 cluster "; cat "$ARTIFACTS/$1-create.log" ; }
-	kubectl config use-context "kind-$1"
-        kubectl get nodes > "$ARTIFACTS/$1-nodes.log" || true
-        kubectl describe pods -n kube-system > "$ARTIFACTS/$1-system-pods.log" || true
+    prepare_kubeconfig "$1" "$3"
+
+    $KIND create cluster --name "$1" --image "$E2E_KIND_VERSION" --config "$2" --kubeconfig="$3" --wait 1m -v 5  > "$ARTIFACTS/$1-create.log" 2>&1 \
+    ||  { echo "unable to start the $1 cluster "; cat "$ARTIFACTS/$1-create.log" ; }
+ 
+    kubectl config --kubeconfig="$3" use-context "kind-$1"
+    kubectl get nodes --kubeconfig="$3" > "$ARTIFACTS/$1-nodes.log" || true
+    kubectl describe pods --kubeconfig="$3" -n kube-system > "$ARTIFACTS/$1-system-pods.log" || true
 }
 
 function prepare_docker_images {
@@ -128,6 +134,40 @@ function cluster_kind_load {
     cluster_kind_load_image "$1" "$IMAGE_TAG"
 }
 
+# $1 cluster
+# $2 kubeconfig
+function kind_load {
+    kubectl config --kubeconfig="$2" use-context "kind-$1"
+
+    if [ "$CREATE_KIND_CLUSTER" == 'true' ]; then
+	    cluster_kind_load "$1"
+    fi
+    if [[ -n ${APPWRAPPER_VERSION:-} ]]; then
+        install_appwrapper "$1" "$2"
+    fi
+    if [[ -n ${JOBSET_VERSION:-} ]]; then
+        install_jobset "$1" "$2"
+    fi
+    if [[ -n ${KUBEFLOW_VERSION:-} ]]; then
+        # 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
+        # 2. Training-operator deployment is modified to enable all kubeflow jobs except for mpi -  https://github.com/kubeflow/training-operator/issues/1777
+        install_kubeflow "$1" "$2"
+    fi
+    if [[ -n ${KUBEFLOW_MPI_VERSION:-} ]]; then
+        install_mpi "$1" "$2"
+    fi
+    if [[ -n ${LEADERWORKERSET_VERSION:-} ]]; then
+        install_lws "$1" "$2"
+    fi
+    if [[ -n ${KUBERAY_VERSION:-} ]]; then
+        install_kuberay "$1" "$2"
+    fi
+    if [[ -n ${CERTMANAGER_VERSION:-} ]]; then
+        install_cert_manager "$2"
+    fi
+}
+
 # $1 cluster
 # $2 image
 function cluster_kind_load_image {
@@ -144,6 +184,7 @@ function cluster_kind_load_image {
     fi
 }
 
+# $1 kubeconfig
 function deploy_with_certmanager() {
     local crd_kust="${ROOT_DIR}/config/components/crd/kustomization.yaml"
     local default_kust="${ROOT_DIR}/config/default/kustomization.yaml"
@@ -167,74 +208,73 @@ function deploy_with_certmanager() {
         $KUSTOMIZE edit add patch --path "cert_metrics_manager_patch.yaml" --kind Deployment
         
         $KUSTOMIZE build "${ROOT_DIR}/test/e2e/config/certmanager" | \
-            kubectl apply --server-side -f -
+        kubectl apply --kubeconfig="$1" --server-side -f -
     )
 
     printf "%s\n" "$crd_backup" > "$crd_kust"
     printf "%s\n" "$default_backup" > "$default_kust"
 }
 
-# $1 cluster
+# $1 kubeconfig
 function cluster_kueue_deploy {
-    kubectl config use-context "kind-${1}"
-    
     if [[ -n ${CERTMANAGER_VERSION:-} ]]; then
         kubectl -n cert-manager wait --for condition=ready pod \
             -l app.kubernetes.io/instance=cert-manager \
             --timeout=5m
         
-        deploy_with_certmanager
+        deploy_with_certmanager "$1"
     else
-        kubectl apply --server-side -k test/e2e/config/default
+        kubectl apply --kubeconfig="$1" --server-side -k "${ROOT_DIR}/test/e2e/config/default"
     fi
 }
 
-#$1 - cluster name
+# $1 cluster name
+# $2 kubeconfig option
 function install_appwrapper {
     cluster_kind_load_image "${1}" "${APPWRAPPER_IMAGE}"
-    kubectl config use-context "kind-${1}"
-    kubectl apply -k "${APPWRAPPER_MANIFEST}"
+    kubectl apply --kubeconfig="$2" --server-side -k "${APPWRAPPER_MANIFEST}"
 }
 
-#$1 - cluster name
+# $1 cluster name
+# $2 kubeconfig option
 function install_jobset {
     cluster_kind_load_image "${1}" "${JOBSET_IMAGE}"
-    kubectl config use-context "kind-${1}"
-    kubectl apply --server-side -f "${JOBSET_MANIFEST}"
+    kubectl apply --kubeconfig="$2" --server-side -f "${JOBSET_MANIFEST}"
 }
 
-#$1 - cluster name
+# $1 cluster name
+# $2 kubeconfig option
 function install_kubeflow {
     cluster_kind_load_image "${1}" "${KUBEFLOW_IMAGE}"
-    kubectl config use-context "kind-${1}"
-    kubectl apply --server-side -k "${KUBEFLOW_MANIFEST_PATCHED}"
+    kubectl apply --kubeconfig="$2" --server-side -k "${KUBEFLOW_MANIFEST_PATCHED}"
 }
 
-#$1 - cluster name
+# $1 cluster name
+# $2 kubeconfig option
 function install_mpi {
     cluster_kind_load_image "${1}" "${KUBEFLOW_MPI_IMAGE/#v}"
-    kubectl config use-context "kind-${1}"
-    kubectl apply --server-side -f "${KUBEFLOW_MPI_MANIFEST}"
+    kubectl apply --kubeconfig="$2" --server-side -f "${KUBEFLOW_MPI_MANIFEST}"
 }
 
-#$1 - cluster name
+# $1 cluster name
+# $2 kubeconfig option
 function install_kuberay {
     cluster_kind_load_image "${1}" "${KUBERAY_RAY_IMAGE}"
     cluster_kind_load_image "${1}" "${KUBERAY_IMAGE}"
-    kubectl config use-context "kind-${1}"
     # create used instead of apply - https://github.com/ray-project/kuberay/issues/504
-    kubectl create -k "${KUBERAY_MANIFEST}"
+    kubectl create --kubeconfig="$2" -k "${KUBERAY_MANIFEST}"
 }
 
+# $1 cluster name
+# $2 kubeconfig option
 function install_lws {
     cluster_kind_load_image "${1}" "${LEADERWORKERSET_IMAGE/#v}"
-    kubectl config use-context "kind-${1}"
-    kubectl apply --server-side -f "${LEADERWORKERSET_MANIFEST}"
+    kubectl apply --kubeconfig="$2" --server-side -f "${LEADERWORKERSET_MANIFEST}"
 }
 
+# $1 kubeconfig option
 function install_cert_manager {
-    kubectl config use-context "kind-${1}"
-    kubectl apply --server-side -f "${CERTMANAGER_MANIFEST}"
+    kubectl apply --kubeconfig="$1" --server-side -f "${CERTMANAGER_MANIFEST}"
 }
 
 INITIAL_IMAGE=$($YQ '.images[] | select(.name == "controller") | [.newName, .newTag] | join(":")' config/components/manager/kustomization.yaml)
@@ -263,4 +303,21 @@ function determine_kuberay_ray_image {
     fi
 
     export KUBERAY_RAY_IMAGE="${ray_image_to_use}"
-}
\ No newline at end of file
+}
+
+# $1 cluster name
+# $2 kubeconfig file path
+function prepare_kubeconfig {
+    local kind_name=$1
+    local kubeconfig=$2
+    if [[ "$kubeconfig" != "" ]]; then
+        cat <<EOF > "$kubeconfig"
+        apiVersion: v1
+        kind: Config
+        preferences: {}
+EOF
+        kubectl config --kubeconfig="$kubeconfig" set-context "kind-$kind_name" \
+        --cluster="$kind_name" \
+        --user="$kind_name"
+    fi
+}
diff --git a/hack/e2e-kueueviz-backend.sh b/hack/e2e-kueueviz-backend.sh
index 94e055685..78cedc5c4 100755
--- a/hack/e2e-kueueviz-backend.sh
+++ b/hack/e2e-kueueviz-backend.sh
@@ -32,7 +32,7 @@ fi
 cleanup() {
   echo "Cleaning up kueueviz processes"
   kill "${BACKEND_PID}" 
-  cluster_cleanup "${KIND_CLUSTER_NAME}"
+  cluster_cleanup "${KIND_CLUSTER_NAME}" ""
   restore_managers_image
 }
 
@@ -40,12 +40,12 @@ cleanup() {
 trap cleanup EXIT
 
 echo Creating kind cluster "${KIND_CLUSTER_NAME}"
-cluster_create "${KIND_CLUSTER_NAME}" "$SOURCE_DIR/$KIND_CLUSTER_FILE"
+cluster_create "${KIND_CLUSTER_NAME}" "$SOURCE_DIR/$KIND_CLUSTER_FILE" ""
 echo Waiting for kind cluster "${KIND_CLUSTER_NAME}" to start...
 prepare_docker_images
 cluster_kind_load "${KIND_CLUSTER_NAME}"
 (cd config/components/manager && $KUSTOMIZE edit set image controller="$IMAGE_TAG")
-cluster_kueue_deploy "${KIND_CLUSTER_NAME}"
+cluster_kueue_deploy ""
 kubectl wait deploy/kueue-controller-manager -nkueue-system --for=condition=available --timeout=5m
 
 # Deploy KueueViz resources
diff --git a/hack/e2e-kueueviz-local.sh b/hack/e2e-kueueviz-local.sh
index d694c07ea..c430d2005 100755
--- a/hack/e2e-kueueviz-local.sh
+++ b/hack/e2e-kueueviz-local.sh
@@ -25,19 +25,19 @@ source "${SOURCE_DIR}/e2e-common.sh"
 cleanup() {
   echo "Cleaning up kueueviz processes"
   kill $BACKEND_PID $FRONTEND_PID
-  cluster_cleanup "$KIND_CLUSTER_NAME"
+  cluster_cleanup "$KIND_CLUSTER_NAME" ""
   restore_managers_image
 }
 
 # Set trap to clean up on exit
 trap cleanup EXIT
 
-cluster_create "${KIND_CLUSTER_NAME}" "$SOURCE_DIR/$KIND_CLUSTER_FILE"
+cluster_create "${KIND_CLUSTER_NAME}" "$SOURCE_DIR/$KIND_CLUSTER_FILE" ""
 echo Waiting for kind cluster "${KIND_CLUSTER_NAME}" to start...
 prepare_docker_images
 cluster_kind_load "${KIND_CLUSTER_NAME}"
 (cd config/components/manager && $KUSTOMIZE edit set image controller="$IMAGE_TAG")
-cluster_kueue_deploy "${KIND_CLUSTER_NAME}"
+cluster_kueue_deploy ""
 kubectl wait deploy/kueue-controller-manager -nkueue-system --for=condition=available --timeout=5m
 
 # Deploy kueueviz resources
diff --git a/hack/e2e-test.sh b/hack/e2e-test.sh
index c760e2443..48104a291 100755
--- a/hack/e2e-test.sh
+++ b/hack/e2e-test.sh
@@ -30,7 +30,7 @@ function cleanup {
         if [ ! -d "$ARTIFACTS" ]; then
             mkdir -p "$ARTIFACTS"
         fi
-        cluster_cleanup "$KIND_CLUSTER_NAME"
+        cluster_cleanup "$KIND_CLUSTER_NAME" ""
     fi
     #do the image restore here for the case when an error happened during deploy
     restore_managers_image
@@ -42,48 +42,20 @@ function startup {
         if [ ! -d "$ARTIFACTS" ]; then
             mkdir -p "$ARTIFACTS"
         fi
-        cluster_create "$KIND_CLUSTER_NAME"  "$SOURCE_DIR/$KIND_CLUSTER_FILE"
-    fi
-}
-
-function kind_load {
-    prepare_docker_images
-
-    if [ "$CREATE_KIND_CLUSTER" == 'true' ]; then
-	      cluster_kind_load "$KIND_CLUSTER_NAME"
-    fi
-
-    if [[ -n ${APPWRAPPER_VERSION:-} ]]; then
-        install_appwrapper "$KIND_CLUSTER_NAME"
-    fi
-    if [[ -n ${JOBSET_VERSION:-} ]]; then
-        install_jobset "$KIND_CLUSTER_NAME"
-    fi
-    if [[ -n ${KUBEFLOW_VERSION:-} ]]; then
-        install_kubeflow "$KIND_CLUSTER_NAME"
-    fi
-    if [[ -n ${KUBEFLOW_MPI_VERSION:-} ]]; then
-        install_mpi "$KIND_CLUSTER_NAME"
-    fi
-    if [[ -n ${LEADERWORKERSET_VERSION:-} ]]; then
-        install_lws "$KIND_CLUSTER_NAME"
-    fi
-    if [[ -n ${KUBERAY_VERSION:-} ]]; then
-        install_kuberay "$KIND_CLUSTER_NAME"
-    fi
-    if [[ -n ${CERTMANAGER_VERSION:-} ]]; then
-        install_cert_manager "$KIND_CLUSTER_NAME"
+        cluster_create "$KIND_CLUSTER_NAME"  "$SOURCE_DIR/$KIND_CLUSTER_FILE" ""
     fi
 }
 
 function kueue_deploy {
     (cd config/components/manager && $KUSTOMIZE edit set image controller="$IMAGE_TAG")
-    cluster_kueue_deploy "$KIND_CLUSTER_NAME"
+    cluster_kueue_deploy ""
 }
 
 trap cleanup EXIT
-startup
-kind_load
+startup &
+prepare_docker_images
+wait
+kind_load "$KIND_CLUSTER_NAME" ""
 kueue_deploy
 
 if [ "$E2E_RUN_ONLY_ENV" == 'true' ]; then
diff --git a/hack/multikueue-e2e-test.sh b/hack/multikueue-e2e-test.sh
index 86c0680cf..95154d025 100755
--- a/hack/multikueue-e2e-test.sh
+++ b/hack/multikueue-e2e-test.sh
@@ -24,6 +24,10 @@ export MANAGER_KIND_CLUSTER_NAME=${KIND_CLUSTER_NAME}-manager
 export WORKER1_KIND_CLUSTER_NAME=${KIND_CLUSTER_NAME}-worker1
 export WORKER2_KIND_CLUSTER_NAME=${KIND_CLUSTER_NAME}-worker2
 
+export MANAGER_KUBECONFIG="${ARTIFACTS}/kubeconfig-$MANAGER_KIND_CLUSTER_NAME"
+export WORKER1_KUBECONFIG="${ARTIFACTS}/kubeconfig-$WORKER1_KIND_CLUSTER_NAME"
+export WORKER2_KUBECONFIG="${ARTIFACTS}/kubeconfig-$WORKER2_KIND_CLUSTER_NAME"
+
 # shellcheck source=hack/e2e-common.sh
 source "${SOURCE_DIR}/e2e-common.sh"
 
@@ -34,12 +38,13 @@ function cleanup {
             mkdir -p "$ARTIFACTS"
         fi
 
-        cluster_cleanup "$MANAGER_KIND_CLUSTER_NAME"
-        cluster_cleanup "$WORKER1_KIND_CLUSTER_NAME"
-        cluster_cleanup "$WORKER2_KIND_CLUSTER_NAME"
+        cluster_cleanup "$MANAGER_KIND_CLUSTER_NAME" "$MANAGER_KUBECONFIG" &
+        cluster_cleanup "$WORKER1_KIND_CLUSTER_NAME" "$WORKER1_KUBECONFIG" &
+        cluster_cleanup "$WORKER2_KIND_CLUSTER_NAME" "$WORKER2_KUBECONFIG" &
     fi
     #do the image restore here for the case when an error happened during deploy
     restore_managers_image
+    wait
 }
 
 
@@ -50,72 +55,40 @@ function startup {
             mkdir -p "$ARTIFACTS"
         fi
 
-	cp "${SOURCE_DIR}/multikueue/manager-cluster.kind.yaml" "$ARTIFACTS"
+        cp "${SOURCE_DIR}/multikueue/manager-cluster.kind.yaml" "$ARTIFACTS"
 
         #Enable the JobManagedBy feature gates for k8s 1.30+ versions 
         IFS=. read -r -a varr <<< "$KIND_VERSION"
         minor=$(( varr[1] ))        
         if [ "$minor" -ge 30 ]; then
-	    echo "Enable JobManagedBy feature in manager's kind config"
-	    $YQ e -i '.featureGates.JobManagedBy = true' "${ARTIFACTS}/manager-cluster.kind.yaml"
+            echo "Enable JobManagedBy feature in manager's kind config"
+            $YQ e -i '.featureGates.JobManagedBy = true' "${ARTIFACTS}/manager-cluster.kind.yaml"
         fi
 
-        cluster_create "$MANAGER_KIND_CLUSTER_NAME" "${ARTIFACTS}/manager-cluster.kind.yaml"
-        cluster_create "$WORKER1_KIND_CLUSTER_NAME" "$SOURCE_DIR/multikueue/worker-cluster.kind.yaml"
-        cluster_create "$WORKER2_KIND_CLUSTER_NAME" "$SOURCE_DIR/multikueue/worker-cluster.kind.yaml"
-    fi
-}
-
-function kind_load {
-    prepare_docker_images
-
-    if [ "$CREATE_KIND_CLUSTER" == 'true' ]
-    then
-        cluster_kind_load "$MANAGER_KIND_CLUSTER_NAME"
-        cluster_kind_load "$WORKER1_KIND_CLUSTER_NAME"
-        cluster_kind_load "$WORKER2_KIND_CLUSTER_NAME"
+        cluster_create "$MANAGER_KIND_CLUSTER_NAME" "${ARTIFACTS}/manager-cluster.kind.yaml" "$MANAGER_KUBECONFIG" &
+        cluster_create "$WORKER1_KIND_CLUSTER_NAME" "$SOURCE_DIR/multikueue/worker-cluster.kind.yaml" "$WORKER1_KUBECONFIG" &
+        cluster_create "$WORKER2_KIND_CLUSTER_NAME" "$SOURCE_DIR/multikueue/worker-cluster.kind.yaml" "$WORKER2_KUBECONFIG" &
     fi
-
-    # JOBSET SETUP
-    install_jobset "$MANAGER_KIND_CLUSTER_NAME"
-    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
-    # 2. Training-operator deployment is modified to enable all kubeflow jobs except for mpi -  https://github.com/kubeflow/training-operator/issues/1777
-    install_kubeflow "$MANAGER_KIND_CLUSTER_NAME"
-    install_kubeflow "$WORKER1_KIND_CLUSTER_NAME"
-    install_kubeflow "$WORKER2_KIND_CLUSTER_NAME"
-    
-    ## MPI
-    install_mpi "$MANAGER_KIND_CLUSTER_NAME"
-    install_mpi "$WORKER1_KIND_CLUSTER_NAME"
-    install_mpi "$WORKER2_KIND_CLUSTER_NAME"
-
-    ## KUBERAY
-    install_kuberay "$MANAGER_KIND_CLUSTER_NAME"
-    install_kuberay "$WORKER1_KIND_CLUSTER_NAME"
-    install_kuberay "$WORKER2_KIND_CLUSTER_NAME"
 }
 
 function kueue_deploy {
-    (cd config/components/manager && $KUSTOMIZE edit set image controller="$IMAGE_TAG")
+    (cd "${ROOT_DIR}/config/components/manager" && $KUSTOMIZE edit set image controller="$IMAGE_TAG")
 
-    cluster_kueue_deploy "$MANAGER_KIND_CLUSTER_NAME"
-    cluster_kueue_deploy "$WORKER1_KIND_CLUSTER_NAME"
-    cluster_kueue_deploy "$WORKER2_KIND_CLUSTER_NAME"
+    cluster_kueue_deploy "$MANAGER_KUBECONFIG"
+    cluster_kueue_deploy "$WORKER1_KUBECONFIG"
+    cluster_kueue_deploy "$WORKER2_KUBECONFIG"
 }
 
 trap cleanup EXIT
-startup
-kind_load
+startup 
+prepare_docker_images
+wait # for clusters creation
+kind_load "$MANAGER_KIND_CLUSTER_NAME" "$MANAGER_KUBECONFIG" &
+kind_load "$WORKER1_KIND_CLUSTER_NAME" "$WORKER1_KUBECONFIG" &
+kind_load "$WORKER2_KIND_CLUSTER_NAME" "$WORKER2_KUBECONFIG" &
+wait # for libraries installation
+# Merge kubeconfigs to one file to be used in tests now
+export KUBECONFIG="$MANAGER_KUBECONFIG:$WORKER1_KUBECONFIG:$WORKER2_KUBECONFIG"
 kueue_deploy
 
 if [ "$E2E_RUN_ONLY_ENV" == 'true' ]; then
diff --git a/pkg/cache/tas_flavor_snapshot.go b/pkg/cache/tas_flavor_snapshot.go
index 974daba88..6f4450ce8 100644
--- a/pkg/cache/tas_flavor_snapshot.go
+++ b/pkg/cache/tas_flavor_snapshot.go
@@ -69,6 +69,15 @@ type domain struct {
 	// assigned to the given domain.
 	state int32
 
+	// sliceState is a temporary state of the topology domains during the
+	// assignment algorithm that denotes the number of slices that can fit within
+	// that domain.
+	//
+	// For domains that are below the requested topology level the algorithm
+	// assigns 0 to that field as this field makes no sense for lower level
+	// domains.
+	sliceState int32
+
 	// levelValues stores the mapping from domain ID back to the
 	// ordered list of values
 	levelValues []string
@@ -516,15 +525,16 @@ func deleteDomain(currentTopologyAssignment *kueue.TopologyAssignment, nodeToRep
 // Algorithm overview:
 // Phase 1:
 //
-//	determine pod counts for each topology domain. Start at the lowest level
+//	determine pod counts and slice count for each topology domain. Start at the lowest level
 //	and bubble up the numbers to the top level
 //
 // Phase 2:
 //
-//	a) select the domain at requested level with count >= requestedCount
-//	b) traverse the structure down level-by-level optimizing the number of used
-//	  domains at each level
-//	c) build the assignment for the lowest level in the hierarchy
+//	a) sort domains using chosen strategy (i.e. starting from the highest free capacity)
+//	b) select consecutive domains at requested level that can fit the workload
+//	c) traverse the structure down level-by-level optimizing the number of used
+//	domains at each level
+//	d) build the assignment for the lowest level in the hierarchy
 func (s *TASFlavorSnapshot) findTopologyAssignment(
 	tasPodSetRequests TASPodSetRequests,
 	assumedUsage map[utiltas.TopologyDomainID]resources.Requests,
@@ -540,8 +550,17 @@ func (s *TASFlavorSnapshot) findTopologyAssignment(
 	podSetTolerations := info.Tolerations
 	podSetNodeSelectors := info.NodeSelector
 	count := tasPodSetRequests.Count
+
+	// If slice topology is not requested then we can assume that slice is a single pod
+	sliceSize, reason := getSliceSizeWithSinglePodAsDefault(tasPodSetRequests.PodSet.TopologyRequest)
+	if len(reason) > 0 {
+		return nil, reason
+	}
+
 	required := isRequired(tasPodSetRequests.PodSet.TopologyRequest)
 	topologyKey := s.levelKeyWithImpliedFallback(&tasPodSetRequests)
+	sliceTopologyKey := s.sliceLevelKeyWithDefault(&tasPodSetRequests, s.lowestLevel())
+
 	unconstrained := isUnconstrained(tasPodSetRequests.PodSet.TopologyRequest, &tasPodSetRequests)
 	if topologyKey == nil {
 		return nil, "topology level not specified"
@@ -550,6 +569,16 @@ func (s *TASFlavorSnapshot) findTopologyAssignment(
 	if !found {
 		return nil, fmt.Sprintf("no requested topology level: %s", *topologyKey)
 	}
+
+	sliceLevelIdx, found := s.resolveLevelIdx(sliceTopologyKey)
+	if !found {
+		return nil, fmt.Sprintf("no requested topology level for slices: %s", sliceTopologyKey)
+	}
+
+	if levelIdx > sliceLevelIdx {
+		return nil, fmt.Sprintf("podset slice topology %s is above the podset topology %s", sliceTopologyKey, *topologyKey)
+	}
+
 	var selector labels.Selector
 	if s.isLowestLevelNode() {
 		sel, err := labels.ValidatedSelectorFromSet(podSetNodeSelectors)
@@ -560,10 +589,12 @@ func (s *TASFlavorSnapshot) findTopologyAssignment(
 	} else {
 		selector = labels.Everything()
 	}
-	// phase 1 - determine the number of pods which can fit in each topology domain
+	// phase 1 - determine the number of pods and slices which can fit in each topology domain
 	s.fillInCounts(
 		requests,
 		assumedUsage,
+		sliceSize,
+		sliceLevelIdx,
 		simulateEmpty,
 		append(podSetTolerations, s.tolerations...),
 		selector,
@@ -571,8 +602,8 @@ func (s *TASFlavorSnapshot) findTopologyAssignment(
 	)
 
 	// phase 2a: determine the level at which the assignment is done along with
-	// the domains which can accommodate all pods
-	fitLevelIdx, currFitDomain, reason := s.findLevelWithFitDomains(levelIdx, required, count, unconstrained)
+	// the domains which can accommodate all pods/slices
+	fitLevelIdx, currFitDomain, reason := s.findLevelWithFitDomains(levelIdx, required, count, sliceSize, unconstrained)
 	if len(reason) > 0 {
 		return nil, reason
 	}
@@ -580,15 +611,42 @@ func (s *TASFlavorSnapshot) findTopologyAssignment(
 	// phase 2b: traverse the tree down level-by-level optimizing the number of
 	// topology domains at each level
 	// if unconstrained is set, we'll only do it once
-	currFitDomain = s.updateCountsToMinimum(currFitDomain, count, unconstrained)
+	currFitDomain = s.updateSliceCountsToMinimum(currFitDomain, count, sliceSize, unconstrained)
 	for levelIdx := fitLevelIdx; levelIdx+1 < len(s.domainsPerLevel); levelIdx++ {
-		lowerFitDomains := s.lowerLevelDomains(currFitDomain)
-		sortedLowerDomains := s.sortedDomains(lowerFitDomains, unconstrained)
-		currFitDomain = s.updateCountsToMinimum(sortedLowerDomains, count, unconstrained)
+		if levelIdx < sliceLevelIdx {
+			// If we are "above" the requested slice topology level, we're greedily assigning pods/slices to
+			// all domains without checking what we've assigned to parent domains.
+			sortedLowerDomains := s.sortedDomains(s.lowerLevelDomains(currFitDomain), unconstrained)
+			currFitDomain = s.updateSliceCountsToMinimum(sortedLowerDomains, count, sliceSize, unconstrained)
+		} else {
+			// If we are "at" or "below" the requested slice topology level, we have to carefully assign pods
+			// to domains based on what we've assigned to parent domains, that's why we're iterating through
+			// each parent domain and assigning `domain.state` amount of pods its child domains.
+			newCurrFitDomain := make([]*domain, 0)
+			for _, domain := range currFitDomain {
+				sortedLowerDomains := s.sortedDomains(domain.children, unconstrained)
+
+				addCurrFitDomain := s.updateCountsToMinimum(sortedLowerDomains, domain.state, unconstrained)
+				newCurrFitDomain = append(newCurrFitDomain, addCurrFitDomain...)
+			}
+			currFitDomain = newCurrFitDomain
+		}
 	}
 	return s.buildAssignment(currFitDomain), ""
 }
 
+func getSliceSizeWithSinglePodAsDefault(podSetTopologyRequest *kueue.PodSetTopologyRequest) (int32, string) {
+	if podSetTopologyRequest != nil && podSetTopologyRequest.PodSetSliceRequiredTopology != nil {
+		if podSetTopologyRequest.PodSetSliceSize != nil {
+			return *podSetTopologyRequest.PodSetSliceSize, ""
+		} else {
+			return 0, "slice topology requested, but slice size not provided"
+		}
+	}
+
+	return 1, ""
+}
+
 // Merges two topology assignments keeping the lexicographical order of levelValues
 func (s *TASFlavorSnapshot) mergeTopologyAssignments(a, b *kueue.TopologyAssignment) *kueue.TopologyAssignment {
 	nodeLevel := len(s.levelKeys) - 1
@@ -669,27 +727,46 @@ func (s *TASFlavorSnapshot) levelKey(topologyRequest *kueue.PodSetTopologyReques
 	}
 }
 
+func (s *TASFlavorSnapshot) sliceLevelKeyWithDefault(tasRequests *TASPodSetRequests, defaultSliceLevelKey string) string {
+	if tasRequests.PodSet.TopologyRequest != nil && tasRequests.PodSet.TopologyRequest.PodSetSliceRequiredTopology != nil {
+		return *tasRequests.PodSet.TopologyRequest.PodSetSliceRequiredTopology
+	}
+	return defaultSliceLevelKey
+}
+
 func isUnconstrained(tr *kueue.PodSetTopologyRequest, tasRequests *TASPodSetRequests) bool {
 	return (tr != nil && tr.Unconstrained != nil && *tr.Unconstrained) || tasRequests.Implied
 }
 
-// findBestFitDomainIdx finds an index of the first domain with the lowest
+// findBestFitDomain finds an index of the first domain with the lowest
 // value of state, higher or equal than count.
 // If such a domain doesn't exist, it returns 0 as it's an index of the domain with the
 // most available resources
-func findBestFitDomainIdx(domains []*domain, count int32) int {
-	bestFitIdx := 0
-	for i, domain := range domains {
-		if domain.state >= count && domain.state < domains[bestFitIdx].state {
+func findBestFitDomain(domains []*domain, count int32) *domain {
+	bestDomain := domains[0]
+	for _, domain := range domains {
+		if domain.state >= count && domain.state < bestDomain.state {
 			// choose the first occurrence of fitting domains
 			// to make it consecutive with other podSet's
-			bestFitIdx = i
+			bestDomain = domain
 		}
 	}
-	return bestFitIdx
+	return bestDomain
 }
 
-func (s *TASFlavorSnapshot) findLevelWithFitDomains(levelIdx int, required bool, podSetSize int32, unconstrained bool) (int, []*domain, string) {
+func findBestFitDomainForSlices(domains []*domain, sliceCount int32) *domain {
+	bestDomain := domains[0]
+	for _, domain := range domains {
+		if domain.sliceState >= sliceCount && domain.sliceState < bestDomain.sliceState {
+			// choose the first occurrence of fitting domains
+			// to make it consecutive with other podSet's
+			bestDomain = domain
+		}
+	}
+	return bestDomain
+}
+
+func (s *TASFlavorSnapshot) findLevelWithFitDomains(levelIdx int, required bool, podSetSize int32, sliceSize int32, unconstrained bool) (int, []*domain, string) {
 	domains := s.domainsPerLevel[levelIdx]
 	if len(domains) == 0 {
 		return 0, nil, fmt.Sprintf("no topology domains at level: %s", s.levelKeys[levelIdx])
@@ -697,30 +774,33 @@ func (s *TASFlavorSnapshot) findLevelWithFitDomains(levelIdx int, required bool,
 	levelDomains := slices.Collect(maps.Values(domains))
 	sortedDomain := s.sortedDomains(levelDomains, unconstrained)
 	topDomain := sortedDomain[0]
-	if useBestFitAlgorithm(unconstrained) && topDomain.state >= podSetSize {
+
+	sliceCount := podSetSize / sliceSize
+	if useBestFitAlgorithm(unconstrained) && topDomain.sliceState >= sliceCount {
 		// optimize the potentially last domain
-		topDomain = sortedDomain[findBestFitDomainIdx(sortedDomain, podSetSize)]
+		topDomain = findBestFitDomainForSlices(sortedDomain, sliceCount)
 	}
-	if topDomain.state < podSetSize {
+	if topDomain.sliceState < sliceCount {
 		if required {
-			return 0, nil, s.notFitMessage(topDomain.state, podSetSize)
+			return 0, nil, s.notFitMessage(topDomain.sliceState, sliceCount, sliceSize)
 		}
 		if levelIdx > 0 && !unconstrained {
-			return s.findLevelWithFitDomains(levelIdx-1, required, podSetSize, unconstrained)
+			return s.findLevelWithFitDomains(levelIdx-1, required, podSetSize, sliceSize, unconstrained)
 		}
 		results := []*domain{}
-		remainingCount := podSetSize
-		for idx := 0; remainingCount > 0 && idx < len(sortedDomain) && sortedDomain[idx].state > 0; idx++ {
-			offset := 0
-			if useBestFitAlgorithm(unconstrained) && sortedDomain[idx].state >= remainingCount {
+		remainingSliceCount := sliceCount
+		for idx := 0; remainingSliceCount > 0 && idx < len(sortedDomain) && sortedDomain[idx].sliceState > 0; idx++ {
+			domain := sortedDomain[idx]
+			if useBestFitAlgorithm(unconstrained) && sortedDomain[idx].sliceState >= remainingSliceCount {
 				// optimize the last domain
-				offset = findBestFitDomainIdx(sortedDomain[idx:], remainingCount)
+				domain = findBestFitDomainForSlices(sortedDomain[idx:], remainingSliceCount)
 			}
-			results = append(results, sortedDomain[idx+offset])
-			remainingCount -= sortedDomain[idx+offset].state
+			results = append(results, domain)
+
+			remainingSliceCount -= domain.sliceState
 		}
-		if remainingCount > 0 {
-			return 0, nil, s.notFitMessage(podSetSize-remainingCount, podSetSize)
+		if remainingSliceCount > 0 {
+			return 0, nil, s.notFitMessage(sliceCount-remainingSliceCount, sliceCount, sliceSize)
 		}
 		return levelIdx, results, ""
 	}
@@ -738,14 +818,39 @@ func useLeastFreeCapacityAlgorithm(unconstrained bool) bool {
 		(unconstrained && features.Enabled(features.TASProfileMixed))
 }
 
+func (s *TASFlavorSnapshot) updateSliceCountsToMinimum(domains []*domain, count int32, sliceSize int32, unconstrained bool) []*domain {
+	result := make([]*domain, 0)
+	remainingSlices := count / sliceSize
+	for i, domain := range domains {
+		if useBestFitAlgorithm(unconstrained) && domain.sliceState >= remainingSlices {
+			// optimize the last domain
+			domain = findBestFitDomainForSlices(domains[i:], remainingSlices)
+		}
+
+		if domain.sliceState >= remainingSlices {
+			domain.state = remainingSlices * sliceSize
+			domain.sliceState = remainingSlices
+			result = append(result, domain)
+			return result
+		}
+		domain.state = domain.sliceState * sliceSize
+		remainingSlices -= domain.sliceState
+		result = append(result, domain)
+	}
+	s.log.Error(errCodeAssumptionsViolated, "unexpected remainingCount",
+		"remainingSlices", remainingSlices,
+		"count", count,
+		"leaves", s.leaves)
+	return nil
+}
+
 func (s *TASFlavorSnapshot) updateCountsToMinimum(domains []*domain, count int32, unconstrained bool) []*domain {
 	result := make([]*domain, 0)
 	remainingCount := count
 	for i, domain := range domains {
 		if useBestFitAlgorithm(unconstrained) && domain.state >= remainingCount {
 			// optimize the last domain
-			mostAllocatedIdx := findBestFitDomainIdx(domains[i:], remainingCount)
-			domain = domains[i+mostAllocatedIdx]
+			domain = findBestFitDomain(domains[i:], remainingCount)
 		}
 
 		if domain.state >= remainingCount {
@@ -803,22 +908,28 @@ func (s *TASFlavorSnapshot) sortedDomains(domains []*domain, unconstrained bool)
 	isLeastFreeCapacity := useLeastFreeCapacityAlgorithm(unconstrained)
 	result := slices.Clone(domains)
 	slices.SortFunc(result, func(a, b *domain) int {
-		if a.state == b.state {
-			return slices.Compare(a.levelValues, b.levelValues)
+		if a.sliceState != b.sliceState {
+			if isLeastFreeCapacity {
+				// Start from the domain with the least amount of free resources.
+				// Ascending order.
+				return cmp.Compare(a.sliceState, b.sliceState)
+			}
+			return cmp.Compare(b.sliceState, a.sliceState)
 		}
-		if isLeastFreeCapacity {
-			// Start from the domain with the least amount of free resources.
-			// Ascending order.
+
+		if a.state != b.state {
 			return cmp.Compare(a.state, b.state)
 		}
-		// Descending order.
-		return cmp.Compare(b.state, a.state)
+
+		return slices.Compare(a.levelValues, b.levelValues)
 	})
 	return result
 }
 
 func (s *TASFlavorSnapshot) fillInCounts(requests resources.Requests,
 	assumedUsage map[utiltas.TopologyDomainID]resources.Requests,
+	sliceSize int32,
+	sliceLevelIdx int,
 	simulateEmpty bool,
 	tolerations []corev1.Toleration,
 	selector labels.Selector,
@@ -827,6 +938,7 @@ func (s *TASFlavorSnapshot) fillInCounts(requests resources.Requests,
 		// cleanup the state in case some remaining values are present from computing
 		// assignments for previous PodSets.
 		domain.state = 0
+		domain.sliceState = 0
 	}
 	for _, leaf := range s.leaves {
 		// 1. Check Tolerations against Node Taints
@@ -865,7 +977,7 @@ func (s *TASFlavorSnapshot) fillInCounts(requests resources.Requests,
 		leaf.state = requests.CountIn(remainingCapacity)
 	}
 	for _, root := range s.roots {
-		root.state = s.fillInCountsHelper(root)
+		root.state, root.sliceState = s.fillInCountsHelper(root, sliceSize, sliceLevelIdx, 0)
 	}
 }
 
@@ -878,23 +990,44 @@ func belongsToRequiredDomain(leaf *leafDomain, requiredReplacementDomain utiltas
 	return strings.HasPrefix(string(utiltas.DomainID(leaf.levelValues)), string(requiredReplacementDomain))
 }
 
-func (s *TASFlavorSnapshot) fillInCountsHelper(domain *domain) int32 {
+func (s *TASFlavorSnapshot) fillInCountsHelper(domain *domain, sliceSize int32, sliceLevelIdx int, level int) (int32, int32) {
 	// logic for a leaf
 	if len(domain.children) == 0 {
-		return domain.state
+		if level == sliceLevelIdx {
+			// initialize the sliceState if leaf is the request slice level
+			domain.sliceState = domain.state / sliceSize
+		}
+		return domain.state, domain.sliceState
 	}
 	// logic for a parent
 	childrenCapacity := int32(0)
+	sliceCapacity := int32(0)
+
 	for _, child := range domain.children {
-		childrenCapacity += s.fillInCountsHelper(child)
+		addChildrenCapacity, addChildrenSliceCapacity := s.fillInCountsHelper(child, sliceSize, sliceLevelIdx, level+1)
+		childrenCapacity += addChildrenCapacity
+		sliceCapacity += addChildrenSliceCapacity
 	}
 	domain.state = childrenCapacity
-	return childrenCapacity
+	if level == sliceLevelIdx {
+		// initialize the sliceState for the requested slice level.
+		sliceCapacity = domain.state / sliceSize
+	}
+	domain.sliceState = sliceCapacity
+	return childrenCapacity, sliceCapacity
 }
 
-func (s *TASFlavorSnapshot) notFitMessage(fitCount, totalCount int32) string {
-	if fitCount == 0 {
-		return fmt.Sprintf("topology %q doesn't allow to fit any of %v pod(s)", s.topologyName, totalCount)
+func (s *TASFlavorSnapshot) notFitMessage(slicesFitCount, totalRequestesSlicesCount, sliceSize int32) string {
+	if sliceSize == 1 {
+		// each slice is a single pod, so let's refer to them as pods
+		if slicesFitCount == 0 {
+			return fmt.Sprintf("topology %q doesn't allow to fit any of %v pod(s)", s.topologyName, totalRequestesSlicesCount)
+		}
+		return fmt.Sprintf("topology %q allows to fit only %v out of %v pod(s)", s.topologyName, slicesFitCount, totalRequestesSlicesCount)
+	}
+
+	if slicesFitCount == 0 {
+		return fmt.Sprintf("topology %q doesn't allow to fit any of %v slice(s)", s.topologyName, totalRequestesSlicesCount)
 	}
-	return fmt.Sprintf("topology %q allows to fit only %v out of %v pod(s)", s.topologyName, fitCount, totalCount)
+	return fmt.Sprintf("topology %q allows to fit only %v out of %v slice(s)", s.topologyName, slicesFitCount, totalRequestesSlicesCount)
 }
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index 9f45ca351..430b8ccf0 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -676,7 +676,7 @@ func (r *WorkloadReconciler) Create(e event.TypedCreateEvent[*kueue.Workload]) b
 	wlCopy := e.Object.DeepCopy()
 	workload.AdjustResources(ctx, r.client, wlCopy)
 
-	if !workload.HasQuotaReservation(e.Object) {
+	if workload.IsActive(e.Object) && !workload.HasQuotaReservation(e.Object) {
 		if err := r.queues.AddOrUpdateWorkload(wlCopy); err != nil {
 			log.V(2).Info("ignored an error for now", "error", err)
 		}
diff --git a/pkg/controller/jobframework/tas.go b/pkg/controller/jobframework/tas.go
index 6dc389b65..dedf11ca7 100644
--- a/pkg/controller/jobframework/tas.go
+++ b/pkg/controller/jobframework/tas.go
@@ -20,6 +20,7 @@ import (
 	"strconv"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
 
 	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
@@ -53,6 +54,10 @@ func NewPodSetTopologyRequest(meta *metav1.ObjectMeta) *podSetTopologyRequestBui
 	requiredValue, requiredFound := meta.Annotations[kueuealpha.PodSetRequiredTopologyAnnotation]
 	preferredValue, preferredFound := meta.Annotations[kueuealpha.PodSetPreferredTopologyAnnotation]
 	unconstrained, unconstrainedFound := meta.Annotations[kueuealpha.PodSetUnconstrainedTopologyAnnotation]
+
+	sliceRequiredTopologyValue, sliceRequiredTopologyFound := meta.Annotations[kueuealpha.PodSetSliceRequiredTopologyAnnotation]
+	sliceSizeValue, sliceSizeFound := meta.Annotations[kueuealpha.PodSetSliceSizeAnnotation]
+
 	switch {
 	case requiredFound:
 		psTopologyReq.Required = &requiredValue
@@ -65,6 +70,16 @@ func NewPodSetTopologyRequest(meta *metav1.ObjectMeta) *podSetTopologyRequestBui
 		psTopologyReq = nil
 	}
 
+	if sliceRequiredTopologyFound && sliceSizeFound {
+		sliceSizeIntValue, err := strconv.ParseInt(sliceSizeValue, 10, 32)
+		if err != nil {
+			// silently ignore as it should not happen due to earlier validation in a webhook
+		} else {
+			psTopologyReq.PodSetSliceRequiredTopology = &sliceRequiredTopologyValue
+			psTopologyReq.PodSetSliceSize = ptr.To(int32(sliceSizeIntValue))
+		}
+	}
+
 	builder := &podSetTopologyRequestBuilder{request: psTopologyReq}
 	return builder
 }
diff --git a/pkg/controller/jobframework/tas_validation.go b/pkg/controller/jobframework/tas_validation.go
index f52c2cc0c..0a2ce57a6 100644
--- a/pkg/controller/jobframework/tas_validation.go
+++ b/pkg/controller/jobframework/tas_validation.go
@@ -66,10 +66,10 @@ func ValidateTASPodSetRequest(replicaPath *field.Path, replicaMetadata *metav1.O
 
 	// validate slice size annotation
 	if sliceSizeFound {
-		val, err := strconv.Atoi(sliceSizeValue)
+		val, err := strconv.ParseInt(sliceSizeValue, 10, 32)
 		if err != nil {
 			allErrs = append(allErrs, field.Invalid(annotationsPath.Key(kueuealpha.PodSetSliceSizeAnnotation), sliceSizeValue, "must be a numeric value"))
-		} else if val < 1 {
+		} else if int32(val) < 1 {
 			allErrs = append(allErrs, field.Invalid(annotationsPath.Key(kueuealpha.PodSetSliceSizeAnnotation), sliceSizeValue, "must be greater than or equal to 1"))
 		}
 	}
diff --git a/pkg/queue/manager.go b/pkg/queue/manager.go
index 597522adc..f69bbef00 100644
--- a/pkg/queue/manager.go
+++ b/pkg/queue/manager.go
@@ -310,7 +310,7 @@ func (m *Manager) AddLocalQueue(ctx context.Context, q *kueue.LocalQueue) error
 		return fmt.Errorf("listing workloads that match the queue: %w", err)
 	}
 	for _, w := range workloads.Items {
-		if workload.HasQuotaReservation(&w) {
+		if !workload.IsActive(&w) || workload.HasQuotaReservation(&w) {
 			continue
 		}
 		workload.AdjustResources(ctx, m.client, &w)
diff --git a/site/content/en/docs/reference/kueue-config.v1beta1.md b/site/content/en/docs/reference/kueue-config.v1beta1.md
index acfd99ac2..2e9bb4147 100644
--- a/site/content/en/docs/reference/kueue-config.v1beta1.md
+++ b/site/content/en/docs/reference/kueue-config.v1beta1.md
@@ -148,7 +148,7 @@ If the file doesn't exist, default value is kueue-system.</p>
 </td>
 <td>
    <p>ManageJobsWithoutQueueName controls whether or not Kueue reconciles
-jobs that don't set the annotation kueue.x-k8s.io/queue-name.
+jobs that don't set the label kueue.x-k8s.io/queue-name.
 If set to true, then those jobs will be suspended and never started unless
 they are assigned a queue and eventually admitted. This also applies to
 jobs created before starting the kueue controller.



Test patch:
diff --git a/pkg/cache/tas_cache_test.go b/pkg/cache/tas_cache_test.go
index 2ece89b40..0558323af 100644
--- a/pkg/cache/tas_cache_test.go
+++ b/pkg/cache/tas_cache_test.go
@@ -1860,6 +1860,944 @@ func TestFindTopologyAssignment(t *testing.T) {
 				},
 			},
 		},
+		"block required for podset; host required for slices; BestFit": {
+			//        b1
+			//         |
+			//        r1
+			//    /    |    \
+			// x1:3, x2:3, x3:3
+			//
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x2").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x2").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x3").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x3").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(corev1.LabelHostname),
+				PodSetSliceSize:             ptr.To(int32(2)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count: 6,
+			wantAssignment: &kueue.TopologyAssignment{
+				Levels: defaultOneLevel,
+				Domains: []kueue.TopologyDomainAssignment{
+					{
+						Count: 2,
+						Values: []string{
+							"x3",
+						},
+					},
+					{
+						Count: 2,
+						Values: []string{
+							"x2",
+						},
+					},
+					{
+						Count: 2,
+						Values: []string{
+							"x1",
+						},
+					},
+				},
+			},
+		},
+		"block required for podset; host required for slices; prioritize more free slice capacity first and then tight fit; BestFit": {
+			//           b1
+			//            |
+			//           r1
+			//    /    /    \    \
+			// x1:6  x2:5   x3:4  x4:2
+			//
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("6"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x2").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x2").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("5"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x3").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x3").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("4"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x4").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x4").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(corev1.LabelHostname),
+				PodSetSliceSize:             ptr.To(int32(2)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count: 12,
+			wantAssignment: &kueue.TopologyAssignment{
+				Levels: defaultOneLevel,
+				Domains: []kueue.TopologyDomainAssignment{
+					{
+						Count: 6,
+						Values: []string{
+							"x1",
+						},
+					},
+					{
+						Count: 4,
+						Values: []string{
+							"x3",
+						},
+					},
+					{
+						Count: 2,
+						Values: []string{
+							"x4",
+						},
+					},
+				},
+			},
+		},
+		"block required for podset; host required for slices; select domains with tight fit; BestFit": {
+			//        b1
+			//         |
+			//        r1
+			//    /    |    \
+			// x1:3, x2:2, x3:2
+			//
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x2").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x2").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x3").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x3").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(corev1.LabelHostname),
+				PodSetSliceSize:             ptr.To(int32(2)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count: 4,
+			wantAssignment: &kueue.TopologyAssignment{
+				Levels: defaultOneLevel,
+				Domains: []kueue.TopologyDomainAssignment{
+					{
+						Count: 2,
+						Values: []string{
+							"x2",
+						},
+					},
+					{
+						Count: 2,
+						Values: []string{
+							"x3",
+						},
+					},
+				},
+			},
+		},
+		"block required for podset; rack required for slices; BestFit": {
+
+			//        b1            b2
+			//    /       \          |
+			//   r1        r2       r1
+			//  / \     /  / \     /  \
+			// x1 x2  x3 x4  x5   x6   x7
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x2").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x2").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r2-x3").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r2").
+					Label(corev1.LabelHostname, "x3").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r2-x4").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r2").
+					Label(corev1.LabelHostname, "x4").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r2-x5").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r2").
+					Label(corev1.LabelHostname, "x5").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b2-r1-x6").
+					Label(tasBlockLabel, "b2").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x6").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b2-r1-x7").
+					Label(tasBlockLabel, "b2").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x7").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(tasRackLabel),
+				PodSetSliceSize:             ptr.To(int32(2)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count: 4,
+			wantAssignment: &kueue.TopologyAssignment{
+				Levels: defaultOneLevel,
+				Domains: []kueue.TopologyDomainAssignment{
+					{
+						Count: 1,
+						Values: []string{
+							"x1",
+						},
+					},
+					{
+						Count: 1,
+						Values: []string{
+							"x2",
+						},
+					},
+					{
+						Count: 1,
+						Values: []string{
+							"x3",
+						},
+					},
+					{
+						Count: 1,
+						Values: []string{
+							"x4",
+						},
+					},
+				},
+			},
+		},
+		"block preferred for podset; rack required for slices; BestFit": {
+			nodes: defaultNodes,
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Preferred:                   ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(tasRackLabel),
+				PodSetSliceSize:             ptr.To(int32(2)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count: 4,
+			wantAssignment: &kueue.TopologyAssignment{
+				Levels: defaultOneLevel,
+				Domains: []kueue.TopologyDomainAssignment{
+					{
+						Count: 1,
+						Values: []string{
+							"x2",
+						},
+					},
+					{
+						Count: 1,
+						Values: []string{
+							"x3",
+						},
+					},
+					{
+						Count: 2,
+						Values: []string{
+							"x6",
+						},
+					},
+				},
+			},
+		},
+		"block required for podset; host required for slices; optimize last domain; BestFit": {
+			//        b1
+			//         |
+			//        r1
+			//    /    |    \
+			// x1:4, x2:3, x3:2
+			//
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("4"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x2").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x2").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x3").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x3").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(corev1.LabelHostname),
+				PodSetSliceSize:             ptr.To(int32(2)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count: 6,
+			wantAssignment: &kueue.TopologyAssignment{
+				Levels: defaultOneLevel,
+				Domains: []kueue.TopologyDomainAssignment{
+					{
+						Count: 4,
+						Values: []string{
+							"x1",
+						},
+					},
+					{
+						Count: 2,
+						Values: []string{
+							"x3",
+						},
+					},
+				},
+			},
+		},
+		"block required for podset; host required for slices; LeastFreeCapacity": {
+			//        b1
+			//         |
+			//        r1
+			//    /    |    \
+			// x1:4, x2:3, x3:2
+			//
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("4"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x2").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x2").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x3").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x3").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(corev1.LabelHostname),
+				PodSetSliceSize:             ptr.To(int32(2)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count: 4,
+			wantAssignment: &kueue.TopologyAssignment{
+				Levels: defaultOneLevel,
+				Domains: []kueue.TopologyDomainAssignment{
+					{
+						Count: 2,
+						Values: []string{
+							"x2",
+						},
+					},
+					{
+						Count: 2,
+						Values: []string{
+							"x3",
+						},
+					},
+				},
+			},
+			enableFeatureGates: []featuregate.Feature{features.TASProfileLeastFreeCapacity},
+		},
+		"block preferred for podset; host required for slices; LeastFreeCapacity": {
+			//nolint:dupword // suppress duplicate r1 word
+			//        b1                b2
+			//         |                 |
+			//        r1                r1
+			//    /    |    \        /   |
+			// x1:4, x2:3, x3:2   x4:4  x5:2
+			//
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("4"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x2").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x2").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x3").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x3").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b2-r1-x4").
+					Label(tasBlockLabel, "b2").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x4").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("4"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b2-r1-x5").
+					Label(tasBlockLabel, "b2").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x5").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Preferred:                   ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(corev1.LabelHostname),
+				PodSetSliceSize:             ptr.To(int32(2)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count: 8,
+			wantAssignment: &kueue.TopologyAssignment{
+				Levels: defaultOneLevel,
+				Domains: []kueue.TopologyDomainAssignment{
+					{
+						Count: 2,
+						Values: []string{
+							"x2",
+						},
+					},
+					{
+						Count: 2,
+						Values: []string{
+							"x3",
+						},
+					},
+					{
+						Count: 2,
+						Values: []string{
+							"x1",
+						},
+					},
+					{
+						Count: 2,
+						Values: []string{
+							"x5",
+						},
+					},
+				},
+			},
+			enableFeatureGates: []featuregate.Feature{features.TASProfileLeastFreeCapacity},
+		},
+		"block preferred for podset; host required for slices; 2 blocks with unbalanced subdomains; BestFit": {
+			//nolint:dupword // suppress duplicate r1 word
+			//        b1                b2
+			//         |                 |
+			//        r1                r1
+			//    /    |    \            |
+			// x1:3, x2:3, x3:3   		x4:6
+			//
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x2").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x2").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r1-x3").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x3").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b2-r1-x4").
+					Label(tasBlockLabel, "b2").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x4").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("6"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Preferred:                   ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(corev1.LabelHostname),
+				PodSetSliceSize:             ptr.To(int32(3)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count: 12,
+			wantAssignment: &kueue.TopologyAssignment{
+				Levels: defaultOneLevel,
+				Domains: []kueue.TopologyDomainAssignment{
+					{
+						Count: 3,
+						Values: []string{
+							"x1",
+						},
+					},
+					{
+						Count: 3,
+						Values: []string{
+							"x2",
+						},
+					},
+					{
+						Count: 6,
+						Values: []string{
+							"x4",
+						},
+					},
+				},
+			},
+		},
+		"block required for podset; rack required for slices; podset fits in a block, but slices do not fit in racks": {
+
+			//         b1
+			//     /    |    \
+			//   r1    r2    r3
+			//   |      |     |
+			//   x1:2  x2:2  x3:2
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r2-x2").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r2").
+					Label(corev1.LabelHostname, "x2").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r3-x3").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r3").
+					Label(corev1.LabelHostname, "x3").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(tasRackLabel),
+				PodSetSliceSize:             ptr.To(int32(3)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count:      6,
+			wantReason: `topology "default" doesn't allow to fit any of 2 slice(s)`,
+		},
+		"block required for podset; rack required for slices; only 1 out of 2 slices fit the topology": {
+
+			//           b1:6
+			//     /    /    \    \
+			//   r1    r2    r3    r4
+			//   |      |     |     |
+			//   x1:3  x2:1  x3:1  x4:1
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r2-x2").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r2").
+					Label(corev1.LabelHostname, "x2").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r3-x3").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r3").
+					Label(corev1.LabelHostname, "x3").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r4-x4").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r4").
+					Label(corev1.LabelHostname, "x4").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("1"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(tasRackLabel),
+				PodSetSliceSize:             ptr.To(int32(3)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count:      6,
+			wantReason: `topology "default" allows to fit only 1 out of 2 slice(s)`,
+		},
+		"block required for podset; rack required for slices; podset fits in both blocks, but slices fit in only one block": {
+
+			//       b1:6          b2:6
+			//    /    |    \      |    \
+			//   r1    r2    r3    r4    r5
+			//    |    |     |     |     |
+			//   x1:2  x2:2  x3:2  x4:3  x5:3
+			nodes: []corev1.Node{
+				*testingnode.MakeNode("b1-r1-x1").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r1").
+					Label(corev1.LabelHostname, "x1").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r2-x2").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r2").
+					Label(corev1.LabelHostname, "x2").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b1-r3-x3").
+					Label(tasBlockLabel, "b1").
+					Label(tasRackLabel, "r3").
+					Label(corev1.LabelHostname, "x3").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("2"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b2-r4-x4").
+					Label(tasBlockLabel, "b2").
+					Label(tasRackLabel, "r4").
+					Label(corev1.LabelHostname, "x4").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+				*testingnode.MakeNode("b2-r5-x5").
+					Label(tasBlockLabel, "b2").
+					Label(tasRackLabel, "r5").
+					Label(corev1.LabelHostname, "x5").
+					StatusAllocatable(corev1.ResourceList{
+						corev1.ResourceCPU:  resource.MustParse("3"),
+						corev1.ResourcePods: resource.MustParse("10"),
+					}).
+					Ready().
+					Obj(),
+			},
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(tasRackLabel),
+				PodSetSliceSize:             ptr.To(int32(3)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count: 6,
+			wantAssignment: &kueue.TopologyAssignment{
+				Levels: defaultOneLevel,
+				Domains: []kueue.TopologyDomainAssignment{
+					{
+						Count: 3,
+						Values: []string{
+							"x4",
+						},
+					},
+					{
+						Count: 3,
+						Values: []string{
+							"x5",
+						},
+					},
+				},
+			},
+		},
+		"slice required topology level cannot be above the main required topology level": {
+			nodes: defaultNodes,
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(corev1.LabelHostname),
+				PodSetSliceRequiredTopology: ptr.To(tasBlockLabel),
+				PodSetSliceSize:             ptr.To(int32(1)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count:      1,
+			wantReason: "podset slice topology cloud.com/topology-block is above the podset topology kubernetes.io/hostname",
+		},
+		"slice size is required when slice topology is requested": {
+			nodes: defaultNodes,
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(tasBlockLabel),
+				PodSetSliceRequiredTopology: ptr.To(corev1.LabelHostname),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count:      1,
+			wantReason: "slice topology requested, but slice size not provided",
+		},
+		"cannot request not existing slice topology": {
+			nodes: defaultNodes,
+			topologyRequest: &kueue.PodSetTopologyRequest{
+				Required:                    ptr.To(string(tasBlockLabel)),
+				PodSetSliceRequiredTopology: ptr.To("not-existing-topology-level"),
+				PodSetSliceSize:             ptr.To(int32(1)),
+			},
+			levels: defaultThreeLevels,
+			requests: resources.Requests{
+				corev1.ResourceCPU: 1000,
+			},
+			count:      1,
+			wantReason: "no requested topology level for slices: not-existing-topology-level",
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
diff --git a/test/e2e/customconfigs/objectretentionpolicies_test.go b/test/e2e/customconfigs/objectretentionpolicies_test.go
index 5774b4a63..c94929281 100644
--- a/test/e2e/customconfigs/objectretentionpolicies_test.go
+++ b/test/e2e/customconfigs/objectretentionpolicies_test.go
@@ -35,6 +35,102 @@ import (
 	"sigs.k8s.io/kueue/test/util"
 )
 
+var _ = ginkgo.Describe("ObjectRetentionPolicies", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
+	var (
+		ns *corev1.Namespace
+		rf *kueue.ResourceFlavor
+		cq *kueue.ClusterQueue
+		lq *kueue.LocalQueue
+	)
+
+	ginkgo.BeforeEach(func() {
+		ns = util.CreateNamespaceFromPrefixWithLog(ctx, k8sClient, "orp-")
+
+		rf = testing.MakeResourceFlavor("default").Obj()
+		gomega.Expect(k8sClient.Create(ctx, rf)).Should(gomega.Succeed())
+
+		cq = testing.MakeClusterQueue("cq").
+			ResourceGroup(*testing.MakeFlavorQuotas(rf.Name).Resource(corev1.ResourceCPU, "10").Obj()).
+			Obj()
+		gomega.Expect(k8sClient.Create(ctx, cq)).Should(gomega.Succeed())
+
+		lq = testing.MakeLocalQueue("lq", ns.Name).ClusterQueue(cq.Name).Obj()
+		gomega.Expect(k8sClient.Create(ctx, lq)).Should(gomega.Succeed())
+	})
+
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, cq, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, rf, true)
+	})
+
+	ginkgo.It("should delete the Workload after enabling the ObjectRetentionPolicies feature gate", func() {
+		waitForPodsReady := &configapi.WaitForPodsReady{
+			Enable:          true,
+			BlockAdmission:  ptr.To(true),
+			Timeout:         &metav1.Duration{Duration: util.TinyTimeout},
+			RecoveryTimeout: nil,
+			RequeuingStrategy: &configapi.RequeuingStrategy{
+				Timestamp:          ptr.To(configapi.EvictionTimestamp),
+				BackoffBaseSeconds: ptr.To(int32(1)),
+				BackoffLimitCount:  ptr.To(int32(1)),
+			},
+		}
+
+		updateKueueConfiguration(func(cfg *configapi.Configuration) {
+			cfg.FeatureGates = nil
+			cfg.ObjectRetentionPolicies = nil
+			cfg.WaitForPodsReady = waitForPodsReady.DeepCopy()
+		})
+
+		job := testingjob.MakeJob("job", ns.Name).
+			Queue(kueue.LocalQueueName(lq.Name)).
+			RequestAndLimit(corev1.ResourceCPU, "1").
+			Obj()
+		ginkgo.By("Creating a Job", func() {
+			util.MustCreate(ctx, k8sClient, job)
+		})
+
+		wlKey := types.NamespacedName{
+			Namespace: job.Namespace,
+			Name:      workloadjob.GetWorkloadNameForJob(job.Name, job.UID),
+		}
+		wl := &kueue.Workload{}
+
+		ginkgo.By("Waiting for the Workload to be deactivated", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+				g.Expect(wl.Spec.Active).To(gomega.Equal(ptr.To(false)))
+			}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("Enable ObjectRetentionPolicies feature gate", func() {
+			updateKueueConfiguration(func(cfg *configapi.Configuration) {
+				cfg.FeatureGates = map[string]bool{string(features.ObjectRetentionPolicies): true}
+				cfg.ObjectRetentionPolicies = &configapi.ObjectRetentionPolicies{
+					Workloads: &configapi.WorkloadRetentionPolicy{
+						AfterDeactivatedByKueue: &metav1.Duration{Duration: util.TinyTimeout},
+					},
+				}
+				cfg.WaitForPodsReady = waitForPodsReady.DeepCopy()
+			})
+		})
+
+		ginkgo.By("Checking that the Job is deleted", func() {
+			createdJob := &batchv1.Job{}
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(job), createdJob)).To(testing.BeNotFoundError())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("Checking that the Workload is deleted", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(testing.BeNotFoundError())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+	})
+})
+
 var _ = ginkgo.Describe("ObjectRetentionPolicies with TinyTimeout", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
 	var (
 		ns *corev1.Namespace
diff --git a/test/e2e/tas/jobset_test.go b/test/e2e/tas/jobset_test.go
index 2a789f078..6cd9dad42 100644
--- a/test/e2e/tas/jobset_test.go
+++ b/test/e2e/tas/jobset_test.go
@@ -145,6 +145,71 @@ var _ = ginkgo.Describe("TopologyAwareScheduling for JobSet", func() {
 				gomega.Expect(wantAssignment).Should(gomega.BeComparableTo(gotAssignment))
 			})
 		})
+
+		ginkgo.It("Should place pods in podset slices based on the ranks-ordering", func() {
+			replicas := 2
+			parallelism := 3
+			numPods := replicas * parallelism
+			sampleJob := testingjobset.MakeJobSet("ranks-jobset", ns.Name).
+				Queue(localQueue.Name).
+				ReplicatedJobs(
+					testingjobset.ReplicatedJobRequirements{
+						Name:        "replicated-job-1",
+						Image:       util.E2eTestAgnHostImage,
+						Args:        util.BehaviorExitFast,
+						Replicas:    int32(replicas),
+						Parallelism: int32(parallelism),
+						Completions: int32(parallelism),
+						PodAnnotations: map[string]string{
+							kueuealpha.PodSetPreferredTopologyAnnotation:     testing.DefaultBlockTopologyLevel,
+							kueuealpha.PodSetSliceRequiredTopologyAnnotation: testing.DefaultBlockTopologyLevel,
+							kueuealpha.PodSetSliceSizeAnnotation:             "3",
+						},
+					},
+				).
+				RequestAndLimit("replicated-job-1", extraResource, "1").
+				Obj()
+			util.MustCreate(ctx, k8sClient, sampleJob)
+
+			ginkgo.By("JobSet is unsuspended", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(sampleJob), sampleJob)).To(gomega.Succeed())
+					g.Expect(sampleJob.Spec.Suspend).Should(gomega.Equal(ptr.To(false)))
+				}, util.LongTimeout, 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())
+			})
+
+			ginkgo.By("verify the assignment of pods are as expected with rank-based ordering", func() {
+				gomega.Expect(k8sClient.List(ctx, pods, client.InNamespace(ns.Name))).To(gomega.Succeed())
+				gotAssignment := readRankAssignmentsFromJobSetPods(pods.Items)
+				wantAssignment := map[string]string{
+					"0/0": "kind-worker",
+					"0/1": "kind-worker2",
+					"0/2": "kind-worker3",
+					"1/0": "kind-worker5",
+					"1/1": "kind-worker6",
+					"1/2": "kind-worker7",
+				}
+				gomega.Expect(wantAssignment).Should(gomega.BeComparableTo(gotAssignment))
+			})
+		})
 	})
 })
 
diff --git a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
index 60088c2ee..743300916 100644
--- a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
@@ -1283,13 +1283,27 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 		podsCountClusterQ   *kueue.ClusterQueue
 		prodLocalQ          *kueue.LocalQueue
 		devLocalQ           *kueue.LocalQueue
+		podsLocalQ          *kueue.LocalQueue
 	)
 
-	ginkgo.BeforeAll(func() {
+	startManager := func() {
 		fwk.StartManager(ctx, cfg, managerAndControllersSetup(false, true, nil))
+	}
+
+	stopManager := func() {
+		fwk.StopManager(ctx)
+	}
+
+	restartManager := func() {
+		stopManager()
+		startManager()
+	}
+
+	ginkgo.BeforeAll(func() {
+		startManager()
 	})
 	ginkgo.AfterAll(func() {
-		fwk.StopManager(ctx)
+		stopManager()
 	})
 
 	ginkgo.BeforeEach(func() {
@@ -1332,12 +1346,22 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 			}).
 			Obj()
 		util.MustCreate(ctx, k8sClient, devClusterQ)
+
 		podsCountClusterQ = testing.MakeClusterQueue("pods-clusterqueue").
 			ResourceGroup(
 				*testing.MakeFlavorQuotas("on-demand").Resource(corev1.ResourcePods, "5").Obj(),
 			).
 			Obj()
 		util.MustCreate(ctx, k8sClient, podsCountClusterQ)
+
+		prodLocalQ = testing.MakeLocalQueue("prod-queue", ns.Name).ClusterQueue(prodClusterQ.Name).Obj()
+		util.MustCreate(ctx, k8sClient, prodLocalQ)
+
+		devLocalQ = testing.MakeLocalQueue("dev-queue", ns.Name).ClusterQueue(devClusterQ.Name).Obj()
+		util.MustCreate(ctx, k8sClient, devLocalQ)
+
+		podsLocalQ = testing.MakeLocalQueue("pods-queue", ns.Name).ClusterQueue(podsCountClusterQ.Name).Obj()
+		util.MustCreate(ctx, k8sClient, podsLocalQ)
 	})
 
 	ginkgo.AfterEach(func() {
@@ -1351,12 +1375,6 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 	})
 
 	ginkgo.It("Should schedule jobs as they fit in their ClusterQueue", framework.SlowSpec, func() {
-		ginkgo.By("creating localQueues")
-		prodLocalQ = testing.MakeLocalQueue("prod-queue", ns.Name).ClusterQueue(prodClusterQ.Name).Obj()
-		util.MustCreate(ctx, k8sClient, prodLocalQ)
-		devLocalQ = testing.MakeLocalQueue("dev-queue", ns.Name).ClusterQueue(devClusterQ.Name).Obj()
-		util.MustCreate(ctx, k8sClient, devLocalQ)
-
 		ginkgo.By("checking the first prod job starts")
 		prodJob1 := testingjob.MakeJob("prod-job1", ns.Name).Queue(kueue.LocalQueueName(prodLocalQ.Name)).Request(corev1.ResourceCPU, "2").Obj()
 		util.MustCreate(ctx, k8sClient, prodJob1)
@@ -1433,11 +1451,11 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 		}()
 
 		ginkgo.By("create a localQueue located in a different namespace as the job")
-		localQueue := testing.MakeLocalQueue("local-queue", ns2.Name).Obj()
-		localQueue.Spec.ClusterQueue = kueue.ClusterQueueReference(prodClusterQ.Name)
+		ns2LocalQ := testing.MakeLocalQueue("local-queue", ns2.Name).Obj()
+		ns2LocalQ.Spec.ClusterQueue = kueue.ClusterQueueReference(prodClusterQ.Name)
 
 		ginkgo.By("create a job")
-		prodJob := testingjob.MakeJob("prod-job", ns.Name).Queue(kueue.LocalQueueName(localQueue.Name)).Request(corev1.ResourceCPU, "2").Obj()
+		prodJob := testingjob.MakeJob("prod-job", ns.Name).Queue(kueue.LocalQueueName(ns2LocalQ.Name)).Request(corev1.ResourceCPU, "2").Obj()
 		util.MustCreate(ctx, k8sClient, prodJob)
 
 		ginkgo.By("job should be suspend")
@@ -1449,8 +1467,8 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 		}, util.Timeout, util.Interval).Should(gomega.Succeed())
 
 		ginkgo.By("creating another localQueue of the same name and in the same namespace as the job")
-		prodLocalQ = testing.MakeLocalQueue(localQueue.Name, ns.Name).ClusterQueue(prodClusterQ.Name).Obj()
-		util.MustCreate(ctx, k8sClient, prodLocalQ)
+		localQ := testing.MakeLocalQueue(ns2LocalQ.Name, ns.Name).ClusterQueue(prodClusterQ.Name).Obj()
+		util.MustCreate(ctx, k8sClient, localQ)
 
 		ginkgo.By("job should be unsuspended and NodeSelector properly set")
 		gomega.Eventually(func(g gomega.Gomega) {
@@ -1635,11 +1653,6 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 	})
 
 	ginkgo.It("Should allow reclaim of resources that are no longer needed", func() {
-		ginkgo.By("creating localQueue", func() {
-			prodLocalQ = testing.MakeLocalQueue("prod-queue", ns.Name).ClusterQueue(prodClusterQ.Name).Obj()
-			util.MustCreate(ctx, k8sClient, prodLocalQ)
-		})
-
 		job1 := testingjob.MakeJob("job1", ns.Name).Queue(kueue.LocalQueueName(prodLocalQ.Name)).
 			Request(corev1.ResourceCPU, "2").
 			Completions(5).
@@ -1702,9 +1715,6 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 	})
 
 	ginkgo.It("Should readmit preempted Job with priorityClass in alternative flavor", func() {
-		devLocalQ = testing.MakeLocalQueue("dev-queue", ns.Name).ClusterQueue(devClusterQ.Name).Obj()
-		util.MustCreate(ctx, k8sClient, devLocalQ)
-
 		highPriorityClass := testing.MakePriorityClass("high").PriorityValue(100).Obj()
 		util.MustCreate(ctx, k8sClient, highPriorityClass)
 		ginkgo.DeferCleanup(func() {
@@ -1749,9 +1759,6 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 	})
 
 	ginkgo.It("Should readmit preempted Job with workloadPriorityClass in alternative flavor", func() {
-		devLocalQ = testing.MakeLocalQueue("dev-queue", ns.Name).ClusterQueue(devClusterQ.Name).Obj()
-		util.MustCreate(ctx, k8sClient, devLocalQ)
-
 		highWorkloadPriorityClass := testing.MakeWorkloadPriorityClass("high-workload").PriorityValue(100).Obj()
 		util.MustCreate(ctx, k8sClient, highWorkloadPriorityClass)
 		ginkgo.DeferCleanup(func() {
@@ -1796,7 +1803,6 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 	})
 
 	ginkgo.It("Should schedule jobs with partial admission", func() {
-		prodLocalQ = testing.MakeLocalQueue("prod-queue", ns.Name).ClusterQueue(prodClusterQ.Name).Obj()
 		job1 := testingjob.MakeJob("job1", ns.Name).
 			Queue(kueue.LocalQueueName(prodLocalQ.Name)).
 			Parallelism(5).
@@ -1805,9 +1811,6 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 			Obj()
 		jobKey := types.NamespacedName{Name: job1.Name, Namespace: job1.Namespace}
 
-		ginkgo.By("creating localQueues")
-		util.MustCreate(ctx, k8sClient, prodLocalQ)
-
 		ginkgo.By("creating the job")
 		util.MustCreate(ctx, k8sClient, job1)
 		wlKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(job1.Name, job1.UID), Namespace: job1.Namespace}
@@ -1883,11 +1886,9 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 	})
 
 	ginkgo.It("Should set the flavor's node selectors if the job is admitted by pods count only", func() {
-		localQ := testing.MakeLocalQueue("dev-queue", ns.Name).ClusterQueue(podsCountClusterQ.Name).Obj()
-		util.MustCreate(ctx, k8sClient, localQ)
 		ginkgo.By("Creating a job with no requests, will set the resource flavors selectors when admitted ", func() {
 			job := testingjob.MakeJob("job", ns.Name).
-				Queue(kueue.LocalQueueName(localQ.Name)).
+				Queue(kueue.LocalQueueName(podsLocalQ.Name)).
 				Parallelism(2).
 				Obj()
 			util.MustCreate(ctx, k8sClient, job)
@@ -1896,13 +1897,9 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 			})
 		})
 	})
-	ginkgo.It("Should schedule updated job and update the workload", func() {
-		localQueue := testing.MakeLocalQueue("local-queue", ns.Name).ClusterQueue(prodClusterQ.Name).Obj()
-		ginkgo.By("create a localQueue", func() {
-			util.MustCreate(ctx, k8sClient, localQueue)
-		})
 
-		job := testingjob.MakeJob(jobName, ns.Name).Queue(kueue.LocalQueueName(localQueue.Name)).Request(corev1.ResourceCPU, "3").Parallelism(2).Suspend(false).Obj()
+	ginkgo.It("Should schedule updated job and update the workload", func() {
+		job := testingjob.MakeJob(jobName, ns.Name).Queue(kueue.LocalQueueName(prodLocalQ.Name)).Request(corev1.ResourceCPU, "3").Parallelism(2).Suspend(false).Obj()
 		lookupKey := types.NamespacedName{Name: job.Name, Namespace: job.Namespace}
 		createdJob := &batchv1.Job{}
 
@@ -1945,11 +1942,7 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 
 	ginkgo.When("Suspend a running Job without requeuing through Workload's spec.active field", func() {
 		ginkgo.It("Should not readmit a job to the queue after Active is changed to false", func() {
-			ginkgo.By("creating localQueue")
-			localQueue := testing.MakeLocalQueue("queue", ns.Name).ClusterQueue(prodClusterQ.Name).Obj()
-			util.MustCreate(ctx, k8sClient, localQueue)
-
-			sampleJob := testingjob.MakeJob("job1", ns.Name).Queue(kueue.LocalQueueName(localQueue.Name)).Request(corev1.ResourceCPU, "2").Obj()
+			sampleJob := testingjob.MakeJob("job1", ns.Name).Queue(kueue.LocalQueueName(prodLocalQ.Name)).Request(corev1.ResourceCPU, "2").Obj()
 			lookupKey1 := types.NamespacedName{Name: sampleJob.Name, Namespace: sampleJob.Namespace}
 			wll := &kueue.Workload{}
 
@@ -2005,7 +1998,7 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 
 			ginkgo.By("checking a second job starts after first job is suspended")
-			sampleJob2 := testingjob.MakeJob("job2", ns.Name).Queue(kueue.LocalQueueName(localQueue.Name)).Request(corev1.ResourceCPU, "2").Obj()
+			sampleJob2 := testingjob.MakeJob("job2", ns.Name).Queue(kueue.LocalQueueName(prodLocalQ.Name)).Request(corev1.ResourceCPU, "2").Obj()
 
 			lookupKey2 := types.NamespacedName{Name: sampleJob2.Name, Namespace: sampleJob2.Namespace}
 			wll2 := &kueue.Workload{}
@@ -2088,6 +2081,62 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 			}, util.Timeout, util.Interval).Should(gomega.Succeed())
 		})
 	})
+
+	ginkgo.It("Shouldn't admit deactivated Workload after manager restart", func() {
+		job := testingjob.MakeJob("job", ns.Name).
+			Queue(kueue.LocalQueueName(prodLocalQ.Name)).
+			Request(corev1.ResourceCPU, "2").
+			Obj()
+
+		ginkgo.By("Creating a Job", func() {
+			util.MustCreate(ctx, k8sClient, job)
+		})
+
+		wl := &kueue.Workload{}
+		wlKey := types.NamespacedName{
+			Name:      workloadjob.GetWorkloadNameForJob(job.Name, job.UID),
+			Namespace: job.Namespace,
+		}
+
+		ginkgo.By("Checking that the Workload is admitted", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+				g.Expect(workload.IsAdmitted(wl)).To(gomega.BeTrue())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			util.ExpectAdmittedWorkloadsTotalMetric(prodClusterQ, 1)
+		})
+
+		ginkgo.By("Deactivate the Workload", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+				wl.Spec.Active = ptr.To(false)
+				g.Expect(k8sClient.Update(ctx, wl)).To(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("Checking that the Workload is deactivated and evicted", func() {
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+				g.Expect(workload.IsActive(wl)).To(gomega.BeFalse())
+				g.Expect(workload.IsEvicted(wl)).To(gomega.BeTrue())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+
+		ginkgo.By("Restarting the manager", func() {
+			restartManager()
+		})
+
+		ginkgo.By("Checking that the Workload is not admitted after restart the manager", func() {
+			gomega.Consistently(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+				g.Expect(workload.IsAdmitted(wl)).To(gomega.BeFalse())
+				g.Expect(workload.IsEvicted(wl)).To(gomega.BeTrue())
+				// Using short intervals to make it likely to fail if the conditions flip
+			}, util.ConsistentDuration, util.ShortInterval).Should(gomega.Succeed())
+			// NOTE: controller restart in integration tests does not reset the metrics
+			util.ExpectAdmittedWorkloadsTotalMetric(prodClusterQ, 1)
+		})
+	})
 })
 
 var _ = ginkgo.Describe("Job controller interacting with Workload controller when waitForPodsReady is enabled", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
diff --git a/test/util/constants.go b/test/util/constants.go
index 536066efc..4068d889c 100644
--- a/test/util/constants.go
+++ b/test/util/constants.go
@@ -42,6 +42,7 @@ const (
 	// a change in the deployment status.
 	StartUpTimeout     = 5 * time.Minute
 	ConsistentDuration = time.Second
+	ShortInterval      = 10 * time.Millisecond
 	Interval           = time.Millisecond * 250
 )