public static void validateSinkConfig(SinkConfig sinkConfig, ValidatableFunctionPackage sinkFunction) {
        try {
            ConnectorDefinition defn = sinkFunction.getFunctionMetaData(ConnectorDefinition.class);
            if (defn != null && defn.getSinkConfigClass() != null) {
                Class configClass = Class.forName(defn.getSinkConfigClass(), true, sinkFunction.getClassLoader());
                validateSinkConfig(sinkConfig, configClass);
            }
        } catch (ClassNotFoundException e) {
            throw new IllegalArgumentException("Could not find sink config class", e);
        }
    }
@Test
    public void testValidateConfig() {
        SinkConfig sinkConfig = createSinkConfig();

        // Good config
        sinkConfig.getConfigs().put("configParameter", "Test");
        SinkConfigUtils.validateSinkConfig(sinkConfig, TestSinkConfig.class);

        // Bad config
        sinkConfig.getConfigs().put("configParameter", null);
        Exception e = expectThrows(IllegalArgumentException.class,
                () -> SinkConfigUtils.validateSinkConfig(sinkConfig, TestSinkConfig.class));
        assertTrue(e.getMessage().contains("Could not validate sink config: Field 'configParameter' cannot be null!"));
    }