public void deleteStatefulSet() throws InterruptedException {
        String statefulSetName = createJobName(instanceConfig.getFunctionDetails(), this.jobName);
        final V1DeleteOptions options = new V1DeleteOptions();
        options.setGracePeriodSeconds((long) gracePeriodSeconds);
        options.setPropagationPolicy("Foreground");

        String fqfn = FunctionCommon.getFullyQualifiedName(instanceConfig.getFunctionDetails());
        Actions.Action deleteStatefulSet = Actions.Action.builder()
                .actionName(String.format("Deleting statefulset for function %s", fqfn))
                .numRetries(KubernetesRuntimeFactory.numRetries)
                .sleepBetweenInvocationsMs(KubernetesRuntimeFactory.sleepBetweenRetriesMs)
                .supplier(() -> {
                    Response response;
                    try {
                        // cannot use deleteNamespacedStatefulSet because of bug in kuberenetes
                        // https://github.com/kubernetes-client/java/issues/86
                        response = appsClient.deleteNamespacedStatefulSetCall(
                                statefulSetName,
                                jobNamespace, null, null,
                                gracePeriodSeconds, null, "Foreground",
                                options, null)
                                .execute();
                    } catch (ApiException e) {
                        // if already deleted
                        if (e.getCode() == HTTP_NOT_FOUND) {
                            log.warn("Statefulset for function {} does not exist", fqfn);
                            return Actions.ActionResult.builder().success(true).build();
                        }

                        String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(errorMsg)
                                .build();
                    } catch (IOException e) {
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(e.getMessage())
                                .build();
                    }

                    // if already deleted
                    if (response.code() == HTTP_NOT_FOUND) {
                        log.warn("Statefulset for function {} does not exist", fqfn);
                        return Actions.ActionResult.builder().success(true).build();
                    } else {
                        return Actions.ActionResult.builder()
                                .success(response.isSuccessful())
                                .errorMsg(response.message())
                                .build();
                    }
                })
                .build();


        Actions.Action waitForStatefulSetDeletion = Actions.Action.builder()
                .actionName(String.format("Waiting for StatefulSet deletion to complete deletion of function %s", fqfn))
                // set retry period to be about 2x the graceshutdown time
                .numRetries(KubernetesRuntimeFactory.numRetries * 2)
                .sleepBetweenInvocationsMs(KubernetesRuntimeFactory.sleepBetweenRetriesMs * 2)
                .supplier(() -> {
                    V1StatefulSet response;
                    try {
                        response = appsClient.readNamespacedStatefulSet(statefulSetName, jobNamespace, null);
                    } catch (ApiException e) {
                        // statefulset is gone
                        if (e.getCode() == HTTP_NOT_FOUND) {
                            return Actions.ActionResult.builder().success(true).build();
                        }

                        String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(errorMsg)
                                .build();
                    }
                    return Actions.ActionResult.builder()
                            .success(false)
                            .errorMsg(response.getStatus().toString())
                            .build();
                })
                .build();

        // Need to wait for all pods to die so we can cleanup subscriptions.
        Actions.Action waitForStatefulPodsToTerminate = Actions.Action.builder()
                .actionName(String.format("Waiting for pods for function %s to terminate", fqfn))
                .numRetries(KubernetesRuntimeFactory.numRetries * 2)
                .sleepBetweenInvocationsMs(KubernetesRuntimeFactory.sleepBetweenRetriesMs * 2)
                .supplier(() -> {
                    Map<String, String> validLabels = getLabels(instanceConfig.getFunctionDetails());
                    String labels = String.format("tenant=%s,namespace=%s,name=%s",
                            validLabels.get("tenant"),
                            validLabels.get("namespace"),
                            validLabels.get("name"));

                    V1PodList response;
                    try {
                        response = coreClient.listNamespacedPod(jobNamespace, null, null,
                                null, null, labels,
                                null, null, null, null, null);
                    } catch (ApiException e) {

                        String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(errorMsg)
                                .build();
                    }

                    if (response.getItems().size() > 0) {
                        return Actions.ActionResult.builder()
                                .success(false)
                                .errorMsg(response.getItems().size() + " pods still alive.")
                                .build();
                    } else {
                        return Actions.ActionResult.builder()
                                .success(true)
                                .build();
                    }
                })
                .build();


        AtomicBoolean success = new AtomicBoolean(false);
        Actions.newBuilder()
                .addAction(deleteStatefulSet.toBuilder()
                        .continueOn(true)
                        .build())
                .addAction(waitForStatefulSetDeletion.toBuilder()
                        .continueOn(false)
                        .onSuccess((ignored) -> success.set(true))
                        .build())
                .addAction(deleteStatefulSet.toBuilder()
                        .continueOn(true)
                        .build())
                .addAction(waitForStatefulSetDeletion.toBuilder()
                        .onSuccess((ignored) -> success.set(true))
                        .build())
                .run();

        if (!success.get()) {
            throw new RuntimeException(String.format("Failed to delete statefulset for function %s", fqfn));
        } else {
            // wait for pods to terminate
            Actions.newBuilder()
                    .addAction(waitForStatefulPodsToTerminate)
                    .run();
        }
    }
@Test
    public void testDeleteStatefulSetWithTranslatedKubernetesLabelChars() throws Exception {
        InstanceConfig config = createJavaInstanceConfig(FunctionDetails.Runtime.JAVA, false);
        config.setFunctionDetails(createFunctionDetails(FunctionDetails.Runtime.JAVA, false,
                (fb) -> fb.setTenant("c:tenant").setNamespace("c:ns").setName("c:fn")));

        CoreV1Api coreApi = mock(CoreV1Api.class);
        AppsV1Api appsApi = mock(AppsV1Api.class);

        Call successfulCall = mock(Call.class);
        Response okResponse = mock(Response.class);
        when(okResponse.code()).thenReturn(HttpURLConnection.HTTP_OK);
        when(okResponse.isSuccessful()).thenReturn(true);
        when(okResponse.message()).thenReturn("");
        when(successfulCall.execute()).thenReturn(okResponse);

        final String expectedFunctionNamePrefix = String.format("pf-%s-%s-%s", "c-tenant", "c-ns", "c-fn");

        factory = createKubernetesRuntimeFactory(null, 10, 1.0, 1.0);
        factory.setCoreClient(coreApi);
        factory.setAppsClient(appsApi);

        ArgumentMatcher<String> hasTranslatedFunctionName = (String t) -> t.startsWith(expectedFunctionNamePrefix);

        when(appsApi.deleteNamespacedStatefulSetCall(
                argThat(hasTranslatedFunctionName),
                anyString(), isNull(), isNull(), anyInt(), isNull(), anyString(), any(), isNull())).thenReturn(successfulCall);

        ApiException notFoundException = mock(ApiException.class);
        when(notFoundException.getCode()).thenReturn(HttpURLConnection.HTTP_NOT_FOUND);
        when(appsApi.readNamespacedStatefulSet(
                argThat(hasTranslatedFunctionName), anyString(), isNull())).thenThrow(notFoundException);

        V1PodList podList = mock(V1PodList.class);
        when(podList.getItems()).thenReturn(Collections.emptyList());

        String expectedLabels = String.format("tenant=%s,namespace=%s,name=%s", "c-tenant", "c-ns", "c-fn");

        when(coreApi.listNamespacedPod(anyString(), isNull(), isNull(), isNull(), isNull(),
                eq(expectedLabels), isNull(), isNull(), isNull(), isNull(), isNull())).thenReturn(podList);
        KubernetesRuntime kr = factory.createContainer(config, "/test/code", "code.yml", "/test/transforms", "transform.yml", Long.MIN_VALUE);
        kr.deleteStatefulSet();

        verify(coreApi).listNamespacedPod(anyString(), isNull(), isNull(), isNull(), isNull(),
                eq(expectedLabels), isNull(), isNull(), isNull(), isNull(), isNull());
    }