Repository: quarkusio/quarkus
Issue: Fail to redeploy to Kubernetes

### Describe the bug

When running `mvn clean package -Dquarkus.kubernetes.deploy=true` to update an existing deployment in Kubernetes, it fails with:

```
Build step io.quarkus.kubernetes.deployment.KubernetesDeployer#deploy threw an exception: io.fabric8.kubernetes.client.KubernetesClientException: Failure executing: POST at: https://192.168.64.75:8443/api/v1/namespaces/default/services. Message: Service "my-app" is invalid: spec.ports[0].nodePort: Invalid value: 31084: provided port is already allocated.
```

Ths previously worked in 1.x

### Expected behavior

Image is built and updated deployment rolled out to Kubernetes

### Actual behavior

Exception when running `mvn clean package -Dquarkus.kubernetes.deploy=true` if application already deployed

### How to Reproduce?

_No response_

### Output of `uname -a` or `ver`

_No response_

### Output of `java -version`

_No response_

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

_No response_

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

_No response_

### Additional information

_No response_

Changed files:
extensions/kubernetes/vanilla/deployment/src/main/java/io/quarkus/kubernetes/deployment/KubernetesDeployer.java

diff --git a/extensions/kubernetes/vanilla/deployment/src/main/java/io/quarkus/kubernetes/deployment/KubernetesDeployer.java b/extensions/kubernetes/vanilla/deployment/src/main/java/io/quarkus/kubernetes/deployment/KubernetesDeployer.java
index c358fb29ea5..362b63ae063 100644
--- a/extensions/kubernetes/vanilla/deployment/src/main/java/io/quarkus/kubernetes/deployment/KubernetesDeployer.java
+++ b/extensions/kubernetes/vanilla/deployment/src/main/java/io/quarkus/kubernetes/deployment/KubernetesDeployer.java
@@ -11,11 +11,12 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.nio.file.Path;
-import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
@@ -25,6 +26,7 @@
 import io.dekorate.utils.Serialization;
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.KubernetesList;
+import io.fabric8.kubernetes.api.model.Service;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.openshift.api.model.Route;
@@ -42,7 +44,6 @@
 import io.quarkus.deployment.pkg.builditem.OutputTargetBuildItem;
 import io.quarkus.kubernetes.client.deployment.KubernetesClientErrorHandler;
 import io.quarkus.kubernetes.client.spi.KubernetesClientBuildItem;
-import io.quarkus.kubernetes.spi.KubernetesDeploymentTargetBuildItem;
 
 public class KubernetesDeployer {
 
@@ -60,7 +61,7 @@ public void selectDeploymentTarget(ContainerImageInfoBuildItem containerImageInf
         Optional<String> activeContainerImageCapability = ContainerImageCapabilitiesUtil
                 .getActiveContainerImageCapability(capabilities);
 
-        if (!activeContainerImageCapability.isPresent()) {
+        if (activeContainerImageCapability.isEmpty()) {
             // we can't thrown an exception here, because it could prevent the Kubernetes resources from being generated
             return;
         }
@@ -85,10 +86,10 @@ public void deploy(KubernetesClientBuildItem kubernetesClient,
             return;
         }
 
-        if (!selectedDeploymentTarget.isPresent()) {
+        if (selectedDeploymentTarget.isEmpty()) {
 
-            if (!ContainerImageCapabilitiesUtil
-                    .getActiveContainerImageCapability(capabilities).isPresent()) {
+            if (ContainerImageCapabilitiesUtil
+                    .getActiveContainerImageCapability(capabilities).isEmpty()) {
                 throw new RuntimeException(
                         "A Kubernetes deployment was requested but no extension was found to build a container image. Consider adding one of following extensions: "
                                 + CONTAINER_IMAGE_EXTENSIONS_STR + ".");
@@ -158,11 +159,6 @@ private DeploymentTargetEntry determineDeploymentTarget(
         return selectedTarget;
     }
 
-    private Optional<KubernetesDeploymentTargetBuildItem> getOptionalDeploymentTarget(
-            List<KubernetesDeploymentTargetBuildItem> deploymentTargets, String name) {
-        return deploymentTargets.stream().filter(d -> name.equals(d.getName())).findFirst();
-    }
-
     private DeploymentResultBuildItem deploy(DeploymentTargetEntry deploymentTarget,
             KubernetesClient client, Path outputDir,
             OpenshiftConfig openshiftConfig, ApplicationInfoBuildItem applicationInfo) {
@@ -174,12 +170,13 @@ private DeploymentResultBuildItem deploy(DeploymentTargetEntry deploymentTarget,
 
         try (FileInputStream fis = new FileInputStream(manifest)) {
             KubernetesList list = Serialization.unmarshalAsList(fis);
-            distinct(list.getItems()).forEach(i -> {
-                if (KNATIVE.equals(deploymentTarget.getName().toLowerCase())) {
-                    client.resource(i).inNamespace(namespace).deletingExisting().createOrReplace();
-                } else {
-                    client.resource(i).inNamespace(namespace).createOrReplace();
+            list.getItems().stream().filter(distinctByResourceKey()).forEach(i -> {
+                final var r = client.resource(i).inNamespace(namespace);
+                if (shouldDeleteExisting(deploymentTarget, i)) {
+                    r.delete();
+                    r.waitUntilCondition(Objects::isNull, 10, TimeUnit.SECONDS);
                 }
+                r.createOrReplace();
                 log.info("Applied: " + i.getKind() + " " + i.getMetadata().getName() + ".");
             });
 
@@ -221,13 +218,15 @@ private void printExposeInformation(KubernetesClient client, KubernetesList list
         }
     }
 
-    public static Predicate<HasMetadata> distictByResourceKey() {
+    private static boolean shouldDeleteExisting(DeploymentTargetEntry deploymentTarget, HasMetadata resource) {
+        return KNATIVE.equalsIgnoreCase(deploymentTarget.getName())
+                || resource instanceof Service
+                || (Objects.equals("v1", resource.getApiVersion()) && Objects.equals("Service", resource.getKind()));
+    }
+
+    private static Predicate<HasMetadata> distinctByResourceKey() {
         Map<Object, Boolean> seen = new ConcurrentHashMap<>();
         return t -> seen.putIfAbsent(t.getApiVersion() + "/" + t.getKind() + ":" + t.getMetadata().getName(),
                 Boolean.TRUE) == null;
     }
-
-    private static Collection<HasMetadata> distinct(Collection<HasMetadata> resources) {
-        return resources.stream().filter(distictByResourceKey()).collect(Collectors.toList());
-    }
 }