@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 testUploadAllWithAdditionalArtifacts() throws Exception {
        File jobJar = getFlinkKubernetesJar();
        File addArtifact1 = TestingUtils.getClassFile(DefaultKubernetesArtifactUploader.class);
        File addArtifact2 = TestingUtils.getClassFile(KubernetesUtils.class);
        String localJobUri = "local://" + jobJar.getAbsolutePath();
        String localAddArtUri = "local://" + addArtifact1.getAbsolutePath();
        String nonLocalAddArtUri = "dummyfs://" + addArtifact2.getAbsolutePath();

        config.set(PipelineOptions.JARS, Collections.singletonList(localJobUri));
        config.set(
                ArtifactFetchOptions.ARTIFACT_LIST,
                Arrays.asList(nonLocalAddArtUri, localAddArtUri));
        artifactUploader.uploadAll(config);

        assertJobJarUri(jobJar.getName());

        List<String> additionalArtifactsResult = config.get(ArtifactFetchOptions.ARTIFACT_LIST);
        assertThat(additionalArtifactsResult).hasSize(2);
        assertThat(additionalArtifactsResult)
                .containsExactlyInAnyOrder(
                        nonLocalAddArtUri, "dummyfs:" + tmpDir.resolve(addArtifact1.getName()));
    }