@Override
    public void uploadAll(Configuration config) throws Exception {
        if (!config.get(KubernetesConfigOptions.LOCAL_UPLOAD_ENABLED)) {
            LOG.info(
                    "Local artifact uploading is disabled. Set '{}' to enable.",
                    KubernetesConfigOptions.LOCAL_UPLOAD_ENABLED.key());
            return;
        }

        final String jobUri = upload(config, getJobUri(config));
        updateConfig(config, PipelineOptions.JARS, Collections.singletonList(jobUri));

        final List<String> additionalUris =
                config.getOptional(ArtifactFetchOptions.ARTIFACT_LIST)
                        .orElse(Collections.emptyList());

        final List<String> uploadedAdditionalUris =
                additionalUris.stream()
                        .map(
                                FunctionUtils.uncheckedFunction(
                                        artifactUri -> upload(config, artifactUri)))
                        .collect(Collectors.toList());

        updateConfig(config, ArtifactFetchOptions.ARTIFACT_LIST, uploadedAdditionalUris);
    }
@Test
    void testUploadAllWithOneJobJar() throws Exception {
        File jar = getFlinkKubernetesJar();
        String localUri = "local://" + jar.getAbsolutePath();

        config.set(PipelineOptions.JARS, Collections.singletonList(localUri));
        artifactUploader.uploadAll(config);

        assertJobJarUri(jar.getName());
    }