@Test public void testMergeRuntimeFlags() { SourceConfig sourceConfig = createSourceConfig(); SourceConfig newFunctionConfig = createUpdatedSourceConfig("runtimeFlags", "-Dfoo=bar2"); SourceConfig mergedConfig = SourceConfigUtils.validateUpdate(sourceConfig, newFunctionConfig); assertEquals( mergedConfig.getRuntimeFlags(), "-Dfoo=bar2" ); mergedConfig.setRuntimeFlags(sourceConfig.getRuntimeFlags()); assertEquals( new Gson().toJson(sourceConfig), new Gson().toJson(mergedConfig) ); }
public static SourceConfig validateUpdate(SourceConfig existingConfig, SourceConfig newConfig) { SourceConfig mergedConfig = existingConfig.toBuilder().build(); if (!existingConfig.getTenant().equals(newConfig.getTenant())) { throw new IllegalArgumentException("Tenants differ"); } if (!existingConfig.getNamespace().equals(newConfig.getNamespace())) { throw new IllegalArgumentException("Namespaces differ"); } if (!existingConfig.getName().equals(newConfig.getName())) { throw new IllegalArgumentException("Function Names differ"); } if (!StringUtils.isEmpty(newConfig.getClassName())) { mergedConfig.setClassName(newConfig.getClassName()); } if (!StringUtils.isEmpty(newConfig.getTopicName())) { mergedConfig.setTopicName(newConfig.getTopicName()); } if (!StringUtils.isEmpty(newConfig.getSerdeClassName())) { mergedConfig.setSerdeClassName(newConfig.getSerdeClassName()); } if (!StringUtils.isEmpty(newConfig.getSchemaType())) { mergedConfig.setSchemaType(newConfig.getSchemaType()); } if (newConfig.getConfigs() != null) { mergedConfig.setConfigs(newConfig.getConfigs()); } if (newConfig.getSecrets() != null) { mergedConfig.setSecrets(newConfig.getSecrets()); } if (newConfig.getProcessingGuarantees() != null && !newConfig.getProcessingGuarantees().equals(existingConfig.getProcessingGuarantees())) { throw new IllegalArgumentException("Processing Guarantees cannot be altered"); } if (newConfig.getParallelism() != null) { mergedConfig.setParallelism(newConfig.getParallelism()); } if (newConfig.getResources() != null) { mergedConfig.setResources(ResourceConfigUtils.merge(existingConfig.getResources(), newConfig.getResources())); } if (!StringUtils.isEmpty(newConfig.getArchive())) { mergedConfig.setArchive(newConfig.getArchive()); } if (!StringUtils.isEmpty(newConfig.getRuntimeFlags())) { mergedConfig.setRuntimeFlags(newConfig.getRuntimeFlags()); } if (!StringUtils.isEmpty(newConfig.getCustomRuntimeOptions())) { mergedConfig.setCustomRuntimeOptions(newConfig.getCustomRuntimeOptions()); } if (isBatchSource(existingConfig) != isBatchSource(newConfig)) { throw new IllegalArgumentException("Sources cannot be update between regular sources and batchsource"); } if (newConfig.getBatchSourceConfig() != null) { validateBatchSourceConfigUpdate(existingConfig.getBatchSourceConfig(), newConfig.getBatchSourceConfig()); mergedConfig.setBatchSourceConfig(newConfig.getBatchSourceConfig()); } return mergedConfig; }
SourceConfigUtils { public static SourceConfig validateUpdate(SourceConfig existingConfig, SourceConfig newConfig) { SourceConfig mergedConfig = existingConfig.toBuilder().build(); if (!existingConfig.getTenant().equals(newConfig.getTenant())) { throw new IllegalArgumentException("Tenants differ"); } if (!existingConfig.getNamespace().equals(newConfig.getNamespace())) { throw new IllegalArgumentException("Namespaces differ"); } if (!existingConfig.getName().equals(newConfig.getName())) { throw new IllegalArgumentException("Function Names differ"); } if (!StringUtils.isEmpty(newConfig.getClassName())) { mergedConfig.setClassName(newConfig.getClassName()); } if (!StringUtils.isEmpty(newConfig.getTopicName())) { mergedConfig.setTopicName(newConfig.getTopicName()); } if (!StringUtils.isEmpty(newConfig.getSerdeClassName())) { mergedConfig.setSerdeClassName(newConfig.getSerdeClassName()); } if (!StringUtils.isEmpty(newConfig.getSchemaType())) { mergedConfig.setSchemaType(newConfig.getSchemaType()); } if (newConfig.getConfigs() != null) { mergedConfig.setConfigs(newConfig.getConfigs()); } if (newConfig.getSecrets() != null) { mergedConfig.setSecrets(newConfig.getSecrets()); } if (newConfig.getProcessingGuarantees() != null && !newConfig.getProcessingGuarantees().equals(existingConfig.getProcessingGuarantees())) { throw new IllegalArgumentException("Processing Guarantees cannot be altered"); } if (newConfig.getParallelism() != null) { mergedConfig.setParallelism(newConfig.getParallelism()); } if (newConfig.getResources() != null) { mergedConfig.setResources(ResourceConfigUtils.merge(existingConfig.getResources(), newConfig.getResources())); } if (!StringUtils.isEmpty(newConfig.getArchive())) { mergedConfig.setArchive(newConfig.getArchive()); } if (!StringUtils.isEmpty(newConfig.getRuntimeFlags())) { mergedConfig.setRuntimeFlags(newConfig.getRuntimeFlags()); } if (!StringUtils.isEmpty(newConfig.getCustomRuntimeOptions())) { mergedConfig.setCustomRuntimeOptions(newConfig.getCustomRuntimeOptions()); } if (isBatchSource(existingConfig) != isBatchSource(newConfig)) { throw new IllegalArgumentException("Sources cannot be update between regular sources and batchsource"); } if (newConfig.getBatchSourceConfig() != null) { validateBatchSourceConfigUpdate(existingConfig.getBatchSourceConfig(), newConfig.getBatchSourceConfig()); mergedConfig.setBatchSourceConfig(newConfig.getBatchSourceConfig()); } return mergedConfig; } }
SourceConfigUtils { public static SourceConfig validateUpdate(SourceConfig existingConfig, SourceConfig newConfig) { SourceConfig mergedConfig = existingConfig.toBuilder().build(); if (!existingConfig.getTenant().equals(newConfig.getTenant())) { throw new IllegalArgumentException("Tenants differ"); } if (!existingConfig.getNamespace().equals(newConfig.getNamespace())) { throw new IllegalArgumentException("Namespaces differ"); } if (!existingConfig.getName().equals(newConfig.getName())) { throw new IllegalArgumentException("Function Names differ"); } if (!StringUtils.isEmpty(newConfig.getClassName())) { mergedConfig.setClassName(newConfig.getClassName()); } if (!StringUtils.isEmpty(newConfig.getTopicName())) { mergedConfig.setTopicName(newConfig.getTopicName()); } if (!StringUtils.isEmpty(newConfig.getSerdeClassName())) { mergedConfig.setSerdeClassName(newConfig.getSerdeClassName()); } if (!StringUtils.isEmpty(newConfig.getSchemaType())) { mergedConfig.setSchemaType(newConfig.getSchemaType()); } if (newConfig.getConfigs() != null) { mergedConfig.setConfigs(newConfig.getConfigs()); } if (newConfig.getSecrets() != null) { mergedConfig.setSecrets(newConfig.getSecrets()); } if (newConfig.getProcessingGuarantees() != null && !newConfig.getProcessingGuarantees().equals(existingConfig.getProcessingGuarantees())) { throw new IllegalArgumentException("Processing Guarantees cannot be altered"); } if (newConfig.getParallelism() != null) { mergedConfig.setParallelism(newConfig.getParallelism()); } if (newConfig.getResources() != null) { mergedConfig.setResources(ResourceConfigUtils.merge(existingConfig.getResources(), newConfig.getResources())); } if (!StringUtils.isEmpty(newConfig.getArchive())) { mergedConfig.setArchive(newConfig.getArchive()); } if (!StringUtils.isEmpty(newConfig.getRuntimeFlags())) { mergedConfig.setRuntimeFlags(newConfig.getRuntimeFlags()); } if (!StringUtils.isEmpty(newConfig.getCustomRuntimeOptions())) { mergedConfig.setCustomRuntimeOptions(newConfig.getCustomRuntimeOptions()); } if (isBatchSource(existingConfig) != isBatchSource(newConfig)) { throw new IllegalArgumentException("Sources cannot be update between regular sources and batchsource"); } if (newConfig.getBatchSourceConfig() != null) { validateBatchSourceConfigUpdate(existingConfig.getBatchSourceConfig(), newConfig.getBatchSourceConfig()); mergedConfig.setBatchSourceConfig(newConfig.getBatchSourceConfig()); } return mergedConfig; }  }
SourceConfigUtils { public static SourceConfig validateUpdate(SourceConfig existingConfig, SourceConfig newConfig) { SourceConfig mergedConfig = existingConfig.toBuilder().build(); if (!existingConfig.getTenant().equals(newConfig.getTenant())) { throw new IllegalArgumentException("Tenants differ"); } if (!existingConfig.getNamespace().equals(newConfig.getNamespace())) { throw new IllegalArgumentException("Namespaces differ"); } if (!existingConfig.getName().equals(newConfig.getName())) { throw new IllegalArgumentException("Function Names differ"); } if (!StringUtils.isEmpty(newConfig.getClassName())) { mergedConfig.setClassName(newConfig.getClassName()); } if (!StringUtils.isEmpty(newConfig.getTopicName())) { mergedConfig.setTopicName(newConfig.getTopicName()); } if (!StringUtils.isEmpty(newConfig.getSerdeClassName())) { mergedConfig.setSerdeClassName(newConfig.getSerdeClassName()); } if (!StringUtils.isEmpty(newConfig.getSchemaType())) { mergedConfig.setSchemaType(newConfig.getSchemaType()); } if (newConfig.getConfigs() != null) { mergedConfig.setConfigs(newConfig.getConfigs()); } if (newConfig.getSecrets() != null) { mergedConfig.setSecrets(newConfig.getSecrets()); } if (newConfig.getProcessingGuarantees() != null && !newConfig.getProcessingGuarantees().equals(existingConfig.getProcessingGuarantees())) { throw new IllegalArgumentException("Processing Guarantees cannot be altered"); } if (newConfig.getParallelism() != null) { mergedConfig.setParallelism(newConfig.getParallelism()); } if (newConfig.getResources() != null) { mergedConfig.setResources(ResourceConfigUtils.merge(existingConfig.getResources(), newConfig.getResources())); } if (!StringUtils.isEmpty(newConfig.getArchive())) { mergedConfig.setArchive(newConfig.getArchive()); } if (!StringUtils.isEmpty(newConfig.getRuntimeFlags())) { mergedConfig.setRuntimeFlags(newConfig.getRuntimeFlags()); } if (!StringUtils.isEmpty(newConfig.getCustomRuntimeOptions())) { mergedConfig.setCustomRuntimeOptions(newConfig.getCustomRuntimeOptions()); } if (isBatchSource(existingConfig) != isBatchSource(newConfig)) { throw new IllegalArgumentException("Sources cannot be update between regular sources and batchsource"); } if (newConfig.getBatchSourceConfig() != null) { validateBatchSourceConfigUpdate(existingConfig.getBatchSourceConfig(), newConfig.getBatchSourceConfig()); mergedConfig.setBatchSourceConfig(newConfig.getBatchSourceConfig()); } return mergedConfig; }  static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSourceDetails sourceDetails); static SourceConfig convertFromDetails(FunctionDetails functionDetails); static ExtractedSourceDetails validate(SourceConfig sourceConfig, Path archivePath,
                                                  File sourcePackageFile, String narExtractionDirectory,
                                                  boolean validateConnectorConfig); static SourceConfig validateUpdate(SourceConfig existingConfig, SourceConfig newConfig); static void validateBatchSourceConfig(BatchSourceConfig batchSourceConfig); static Map<String, Object> extractSourceConfig(Function.SourceSpec sourceSpec, String fqfn); static BatchSourceConfig extractBatchSourceConfig(Map<String, Object> configMap); static Map<String, String> computeBatchSourceIntermediateTopicSubscriptions(Function.FunctionDetails details,
                                                                                       String fqfn); static String computeBatchSourceInstanceSubscriptionName(String tenant, String namespace,
                                                                    String sourceName); static TopicName computeBatchSourceIntermediateTopicName(String tenant, String namespace,
                                                                    String sourceName); static boolean isBatchSource(SourceConfig sourceConfig); static void validateBatchSourceConfigUpdate(BatchSourceConfig existingConfig, BatchSourceConfig newConfig); static void validateConnectorConfig(SourceConfig sourceConfig, ClassLoader classLoader); }
SourceConfigUtils { public static SourceConfig validateUpdate(SourceConfig existingConfig, SourceConfig newConfig) { SourceConfig mergedConfig = existingConfig.toBuilder().build(); if (!existingConfig.getTenant().equals(newConfig.getTenant())) { throw new IllegalArgumentException("Tenants differ"); } if (!existingConfig.getNamespace().equals(newConfig.getNamespace())) { throw new IllegalArgumentException("Namespaces differ"); } if (!existingConfig.getName().equals(newConfig.getName())) { throw new IllegalArgumentException("Function Names differ"); } if (!StringUtils.isEmpty(newConfig.getClassName())) { mergedConfig.setClassName(newConfig.getClassName()); } if (!StringUtils.isEmpty(newConfig.getTopicName())) { mergedConfig.setTopicName(newConfig.getTopicName()); } if (!StringUtils.isEmpty(newConfig.getSerdeClassName())) { mergedConfig.setSerdeClassName(newConfig.getSerdeClassName()); } if (!StringUtils.isEmpty(newConfig.getSchemaType())) { mergedConfig.setSchemaType(newConfig.getSchemaType()); } if (newConfig.getConfigs() != null) { mergedConfig.setConfigs(newConfig.getConfigs()); } if (newConfig.getSecrets() != null) { mergedConfig.setSecrets(newConfig.getSecrets()); } if (newConfig.getProcessingGuarantees() != null && !newConfig.getProcessingGuarantees().equals(existingConfig.getProcessingGuarantees())) { throw new IllegalArgumentException("Processing Guarantees cannot be altered"); } if (newConfig.getParallelism() != null) { mergedConfig.setParallelism(newConfig.getParallelism()); } if (newConfig.getResources() != null) { mergedConfig.setResources(ResourceConfigUtils.merge(existingConfig.getResources(), newConfig.getResources())); } if (!StringUtils.isEmpty(newConfig.getArchive())) { mergedConfig.setArchive(newConfig.getArchive()); } if (!StringUtils.isEmpty(newConfig.getRuntimeFlags())) { mergedConfig.setRuntimeFlags(newConfig.getRuntimeFlags()); } if (!StringUtils.isEmpty(newConfig.getCustomRuntimeOptions())) { mergedConfig.setCustomRuntimeOptions(newConfig.getCustomRuntimeOptions()); } if (isBatchSource(existingConfig) != isBatchSource(newConfig)) { throw new IllegalArgumentException("Sources cannot be update between regular sources and batchsource"); } if (newConfig.getBatchSourceConfig() != null) { validateBatchSourceConfigUpdate(existingConfig.getBatchSourceConfig(), newConfig.getBatchSourceConfig()); mergedConfig.setBatchSourceConfig(newConfig.getBatchSourceConfig()); } return mergedConfig; }  static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSourceDetails sourceDetails); static SourceConfig convertFromDetails(FunctionDetails functionDetails); static ExtractedSourceDetails validate(SourceConfig sourceConfig, Path archivePath,
                                                  File sourcePackageFile, String narExtractionDirectory,
                                                  boolean validateConnectorConfig); static SourceConfig validateUpdate(SourceConfig existingConfig, SourceConfig newConfig); static void validateBatchSourceConfig(BatchSourceConfig batchSourceConfig); static Map<String, Object> extractSourceConfig(Function.SourceSpec sourceSpec, String fqfn); static BatchSourceConfig extractBatchSourceConfig(Map<String, Object> configMap); static Map<String, String> computeBatchSourceIntermediateTopicSubscriptions(Function.FunctionDetails details,
                                                                                       String fqfn); static String computeBatchSourceInstanceSubscriptionName(String tenant, String namespace,
                                                                    String sourceName); static TopicName computeBatchSourceIntermediateTopicName(String tenant, String namespace,
                                                                    String sourceName); static boolean isBatchSource(SourceConfig sourceConfig); static void validateBatchSourceConfigUpdate(BatchSourceConfig existingConfig, BatchSourceConfig newConfig); static void validateConnectorConfig(SourceConfig sourceConfig, ClassLoader classLoader);  }