@Test public void testValidateConfig() throws IOException { mockStatic(ConnectorUtils.class); defn = new ConnectorDefinition(); defn.setSourceConfigClass(SourceConfigUtilsTest.TestSourceConfig.class.getName()); PowerMockito.when(ConnectorUtils.getConnectorDefinition(any())).thenReturn(defn); SourceConfig sourceConfig = createSourceConfig(); sourceConfig.getConfigs().put("configParameter", "Test"); SourceConfigUtils.validateConnectorConfig(sourceConfig, Thread.currentThread().getContextClassLoader()); sourceConfig.getConfigs().put("configParameter", null); Exception e = expectThrows(IllegalArgumentException.class, () -> SourceConfigUtils.validateConnectorConfig(sourceConfig, Thread.currentThread().getContextClassLoader())); assertTrue(e.getMessage().contains("Could not validate source config: Field 'configParameter' cannot be null!")); }
public static void validateConnectorConfig(SourceConfig sourceConfig, ClassLoader classLoader) { try { ConnectorDefinition defn = ConnectorUtils.getConnectorDefinition(classLoader); if (defn.getSourceConfigClass() != null) { Class configClass = Class.forName(defn.getSourceConfigClass(), true, classLoader); Object configObject = ObjectMapperFactory.getThreadLocal().convertValue(sourceConfig.getConfigs(), configClass); if (configObject != null) { ConfigValidation.validateConfig(configObject); } } } catch (IOException e) { throw new IllegalArgumentException("Error validating source config", e); } catch (ClassNotFoundException e) { throw new IllegalArgumentException("Could not find source config class"); } catch (IllegalArgumentException e) { throw new IllegalArgumentException("Could not validate source config: " + e.getMessage()); } }
SourceConfigUtils { public static void validateConnectorConfig(SourceConfig sourceConfig, ClassLoader classLoader) { try { ConnectorDefinition defn = ConnectorUtils.getConnectorDefinition(classLoader); if (defn.getSourceConfigClass() != null) { Class configClass = Class.forName(defn.getSourceConfigClass(), true, classLoader); Object configObject = ObjectMapperFactory.getThreadLocal().convertValue(sourceConfig.getConfigs(), configClass); if (configObject != null) { ConfigValidation.validateConfig(configObject); } } } catch (IOException e) { throw new IllegalArgumentException("Error validating source config", e); } catch (ClassNotFoundException e) { throw new IllegalArgumentException("Could not find source config class"); } catch (IllegalArgumentException e) { throw new IllegalArgumentException("Could not validate source config: " + e.getMessage()); } } }
SourceConfigUtils { public static void validateConnectorConfig(SourceConfig sourceConfig, ClassLoader classLoader) { try { ConnectorDefinition defn = ConnectorUtils.getConnectorDefinition(classLoader); if (defn.getSourceConfigClass() != null) { Class configClass = Class.forName(defn.getSourceConfigClass(), true, classLoader); Object configObject = ObjectMapperFactory.getThreadLocal().convertValue(sourceConfig.getConfigs(), configClass); if (configObject != null) { ConfigValidation.validateConfig(configObject); } } } catch (IOException e) { throw new IllegalArgumentException("Error validating source config", e); } catch (ClassNotFoundException e) { throw new IllegalArgumentException("Could not find source config class"); } catch (IllegalArgumentException e) { throw new IllegalArgumentException("Could not validate source config: " + e.getMessage()); } }  }
SourceConfigUtils { public static void validateConnectorConfig(SourceConfig sourceConfig, ClassLoader classLoader) { try { ConnectorDefinition defn = ConnectorUtils.getConnectorDefinition(classLoader); if (defn.getSourceConfigClass() != null) { Class configClass = Class.forName(defn.getSourceConfigClass(), true, classLoader); Object configObject = ObjectMapperFactory.getThreadLocal().convertValue(sourceConfig.getConfigs(), configClass); if (configObject != null) { ConfigValidation.validateConfig(configObject); } } } catch (IOException e) { throw new IllegalArgumentException("Error validating source config", e); } catch (ClassNotFoundException e) { throw new IllegalArgumentException("Could not find source config class"); } catch (IllegalArgumentException e) { throw new IllegalArgumentException("Could not validate source config: " + e.getMessage()); } }  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 void validateConnectorConfig(SourceConfig sourceConfig, ClassLoader classLoader) { try { ConnectorDefinition defn = ConnectorUtils.getConnectorDefinition(classLoader); if (defn.getSourceConfigClass() != null) { Class configClass = Class.forName(defn.getSourceConfigClass(), true, classLoader); Object configObject = ObjectMapperFactory.getThreadLocal().convertValue(sourceConfig.getConfigs(), configClass); if (configObject != null) { ConfigValidation.validateConfig(configObject); } } } catch (IOException e) { throw new IllegalArgumentException("Error validating source config", e); } catch (ClassNotFoundException e) { throw new IllegalArgumentException("Could not find source config class"); } catch (IllegalArgumentException e) { throw new IllegalArgumentException("Could not validate source config: " + e.getMessage()); } }  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);  }