@VisibleForTesting
    V1StatefulSet createStatefulSet() {
        final String jobName = createJobName(instanceConfig.getFunctionDetails(), this.jobName);

        final V1StatefulSet statefulSet = new V1StatefulSet();

        // setup stateful set metadata
        final V1ObjectMeta objectMeta = new V1ObjectMeta();
        objectMeta.name(jobName);
        objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails()));
        // we don't technically need to set this, but it is useful for testing
        objectMeta.setNamespace(jobNamespace);
        statefulSet.metadata(objectMeta);

        // create the stateful set spec
        final V1StatefulSetSpec statefulSetSpec = new V1StatefulSetSpec();
        statefulSetSpec.serviceName(jobName);
        statefulSetSpec.setReplicas(instanceConfig.getFunctionDetails().getParallelism());

        // Parallel pod management tells the StatefulSet controller to launch or terminate
        // all Pods in parallel, and not to wait for Pods to become Running and Ready or completely
        // terminated prior to launching or terminating another Pod.
        statefulSetSpec.setPodManagementPolicy("Parallel");

        // add selector match labels
        // so the we know which pods to manage
        final V1LabelSelector selector = new V1LabelSelector();
        selector.matchLabels(getLabels(instanceConfig.getFunctionDetails()));
        statefulSetSpec.selector(selector);

        // create a pod template
        final V1PodTemplateSpec podTemplateSpec = new V1PodTemplateSpec();

        // set up pod meta
        final V1ObjectMeta templateMetaData = new V1ObjectMeta().labels(getLabels(instanceConfig.getFunctionDetails()));
        templateMetaData.annotations(getPrometheusAnnotations());
        podTemplateSpec.setMetadata(templateMetaData);

        final List<String> command = getExecutorCommand();
        podTemplateSpec.spec(getPodSpec(command,
                instanceConfig.getFunctionDetails().hasResources() ? instanceConfig.getFunctionDetails().getResources()
                        : null));

        statefulSetSpec.setTemplate(podTemplateSpec);

        statefulSet.spec(statefulSetSpec);

        // let the customizer run but ensure it doesn't change the name so we can find it again
        final V1StatefulSet overridden = manifestCustomizer
                .map((customizer) -> customizer.customizeStatefulSet(instanceConfig.getFunctionDetails(), statefulSet))
                .orElse(statefulSet);
        overridden.getMetadata().name(jobName);

        return statefulSet;
    }
@Test
    public void testCustomKubernetesDownloadCommandsWithAuthWithoutAuthSpec() throws Exception {
        InstanceConfig config = createJavaInstanceConfig(FunctionDetails.Runtime.JAVA, false);
        config.setFunctionDetails(createFunctionDetails(FunctionDetails.Runtime.JAVA, false));

        factory = createKubernetesRuntimeFactory(null,
                10, 1.0, 1.0, Optional.empty(), null, wconfig -> {
                    wconfig.setAuthenticationEnabled(true);
                }, AuthenticationConfig.builder()
                        .clientAuthenticationPlugin("com.MyAuth")
                        .clientAuthenticationParameters("{\"authParam1\": \"authParamValue1\"}")
                        .build());

        KubernetesRuntime container = factory.createContainer(config, userJarFile, userJarFile, null, null, 30l);
        V1StatefulSet spec = container.createStatefulSet();
        String expectedDownloadCommand = "pulsar-admin --admin-url " + pulsarAdminUrl
                + " --auth-plugin com.MyAuth --auth-params {\"authParam1\": \"authParamValue1\"}"
                + " functions download "
                + "--tenant " + TEST_TENANT
                + " --namespace " + TEST_NAMESPACE
                + " --name " + TEST_NAME
                + " --destination-file " + pulsarRootDir + "/" + userJarFile;
        String containerCommand = spec.getSpec().getTemplate().getSpec().getContainers().get(0).getCommand().get(2);
        assertTrue(containerCommand.contains(expectedDownloadCommand), "Found:" + containerCommand);
    }