StatefulSetOperator extends AbstractScalableResourceOperator<KubernetesClient, StatefulSet, StatefulSetList, DoneableStatefulSet, RollableScalableResource<StatefulSet, DoneableStatefulSet>> { protected Future<ReconcileResult<StatefulSet>> internalReplace(String namespace, String name, StatefulSet current, StatefulSet desired, boolean cascading) { try { Promise<ReconcileResult<StatefulSet>> promise = Promise.promise(); long pollingIntervalMs = 1_000; long timeoutMs = operationTimeoutMs; operation().inNamespace(namespace).withName(name).withPropagationPolicy(cascading ? DeletionPropagation.FOREGROUND : DeletionPropagation.ORPHAN).withGracePeriod(-1L).delete(); Future<Void> deletedFut = waitFor(namespace, name, "deleted", pollingIntervalMs, timeoutMs, (ignore1, ignore2) -> { StatefulSet sts = get(namespace, name); log.trace("Checking if {} {} in namespace {} has been deleted", resourceKind, name, namespace); return sts == null; }); deletedFut.onComplete(res -> { if (res.succeeded()) { StatefulSet result = operation().inNamespace(namespace).withName(name).create(desired); log.debug("{} {} in namespace {} has been replaced", resourceKind, name, namespace); promise.complete(wasChanged(current, result) ? ReconcileResult.patched(result) : ReconcileResult.noop(result)); } else { promise.fail(res.cause()); } }); return promise.future(); } catch (Exception e) { log.debug("Caught exception while replacing {} {} in namespace {}", resourceKind, name, namespace, e); return Future.failedFuture(e); } } StatefulSetOperator(Vertx vertx, KubernetesClient client, long operationTimeoutMs);  StatefulSetOperator(Vertx vertx, KubernetesClient client, long operationTimeoutMs,
                               PodOperator podOperator, PvcOperator pvcOperator); Future<Void> maybeRollingUpdate(StatefulSet sts, Function<Pod, List<String>> podNeedsRestart); abstract Future<Void> maybeRollingUpdate(StatefulSet sts, Function<Pod, List<String>> podNeedsRestart, Secret clusterCaSecret, Secret coKeySecret); Future<Void> deletePvc(StatefulSet sts, String pvcName); String getPodName(StatefulSet desired, int podId); @SuppressWarnings("deprecation") static int getStsGeneration(StatefulSet resource); @SuppressWarnings("deprecation") static int getPodGeneration(Pod resource); Future<Void> deleteAsync(String namespace, String name, boolean cascading);  }
@Test public void testInternalReplace(VertxTestContext context) { StatefulSet sts1 = new StatefulSetBuilder() .withNewMetadata() .withNamespace(AbstractResourceOperatorTest.NAMESPACE) .withName(AbstractResourceOperatorTest.RESOURCE_NAME) .endMetadata() .withNewSpec() .withReplicas(3) .withNewTemplate() .withNewMetadata() .endMetadata() .endTemplate() .endSpec() .build(); Map<String, Quantity> requests = new HashMap<>(); requests.put("storage", new Quantity("100Gi")); PersistentVolumeClaim pvc = new PersistentVolumeClaimBuilder() .withNewMetadata() .withName("data") .endMetadata() .withNewSpec() .withAccessModes("ReadWriteOnce") .withNewResources() .withRequests(requests) .endResources() .withStorageClassName("gp2") .endSpec() .build(); StatefulSet sts2 = new StatefulSetBuilder() .withNewMetadata() .withNamespace(AbstractResourceOperatorTest.NAMESPACE) .withName(AbstractResourceOperatorTest.RESOURCE_NAME) .endMetadata() .withNewSpec() .withReplicas(3) .withNewTemplate() .withNewMetadata() .endMetadata() .endTemplate() .withVolumeClaimTemplates(pvc) .endSpec() .build(); Deletable mockDeletable = mock(Deletable.class); when(mockDeletable.delete()).thenReturn(Boolean.TRUE); Resource mockERPD = mock(resourceType()); when(mockERPD.withPropagationPolicy(any(DeletionPropagation.class))).thenReturn(mockDeletable); when(mockERPD.withGracePeriod(anyLong())).thenReturn(mockDeletable); Resource mockResource = mock(resourceType()); when(mockResource.get()).thenReturn(sts1); when(mockResource.withPropagationPolicy(eq(DeletionPropagation.ORPHAN))).thenReturn(mockERPD); when(mockResource.create(any(StatefulSet.class))).thenReturn(sts1); PodOperator podOperator = mock(PodOperator.class); when(podOperator.waitFor(anyString(), anyString(), anyLong(), anyLong(), any(BiPredicate.class))).thenReturn(Future.succeededFuture()); when(podOperator.readiness(anyString(), anyString(), anyLong(), anyLong())).thenReturn(Future.succeededFuture()); when(podOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); when(podOperator.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(new PodBuilder().withNewMetadata().withName("my-pod-0").endMetadata().build())); PvcOperator pvcOperator = mock(PvcOperator.class); when(pvcOperator.reconcile(anyString(), anyString(), any())).thenReturn(Future.succeededFuture()); NonNamespaceOperation mockNameable = mock(NonNamespaceOperation.class); when(mockNameable.withName(matches(RESOURCE_NAME))).thenReturn(mockResource); MixedOperation mockCms = mock(MixedOperation.class); when(mockCms.inNamespace(matches(NAMESPACE))).thenReturn(mockNameable); KubernetesClient mockClient = mock(KubernetesClient.class); mocker(mockClient, mockCms); StatefulSetOperator op = new StatefulSetOperator(AbstractResourceOperatorTest.vertx, mockClient, 5_000L, podOperator, pvcOperator) { @Override public Future<Void> maybeRollingUpdate(StatefulSet sts, Function<Pod, List<String>> podNeedsRestart, Secret clusterCaSecret, Secret coKeySecret) { return Future.succeededFuture(); } @Override protected boolean shouldIncrementGeneration(StatefulSetDiff diff) { return true; } @Override public Future<Void> waitFor(String namespace, String name, String logState, long pollIntervalMs, final long timeoutMs, BiPredicate<String, String> predicate) { return Future.succeededFuture(); } }; Checkpoint async = context.checkpoint(); op.reconcile(sts1.getMetadata().getNamespace(), sts1.getMetadata().getName(), sts2) .onComplete(context.succeeding(rrState -> { verify(mockDeletable).delete(); async.flag(); })); }