@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 testInvalidJobJar() {
        String msg = "The 'pipeline.jars' config must contain one JAR.";

        config.set(PipelineOptions.JARS, Collections.emptyList());
        assertThatThrownBy(() -> artifactUploader.uploadAll(config))
                .isInstanceOf(IllegalArgumentException.class)
                .hasMessage(msg);

        config.set(PipelineOptions.JARS, Arrays.asList("a", "b"));
        assertThatThrownBy(() -> artifactUploader.uploadAll(config))
                .isInstanceOf(IllegalArgumentException.class)
                .hasMessage(msg);
    }