You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@pulsar.apache.org by "nicoloboschi (via GitHub)" <gi...@apache.org> on 2023/04/17 09:55:50 UTC

[GitHub] [pulsar] nicoloboschi opened a new pull request, #20116: [improve][fn] Allow unknown fields in connectors config

nicoloboschi opened a new pull request, #20116:
URL: https://github.com/apache/pulsar/pull/20116

### Motivation

Most of the java connectors (sinks and sources) map the configuration to a java class. Pulsar exposes a method to convert the configuration (`IOConfigUtils#loadWithSecrets`) using a jackson mapper. However is up to the connector to decide how to convert the configuration.
Most of the builtin java connectors do not allow unknown fields in their configuration. This is usually a good thing since it alerts the user that something is malconfigured. But this doesn't leave space for connectors' rollbacks.

The scenario which is not possible to cover is:
- Upgrade a builtin connector which has a new config property
- Upgrade existing connectors and use this new property
- For some reasons, the connectors must be rollbacked. In order to let the connector work, it's required to rollback also the existing connectors' configuration removing the new field. This can be very hard to achieve in large clusters.

### Modifications
- Added a new flag in the functions worker `ignoreUnknownConfigFields` (default false) that strips out fields not supported in the current sink/source config. The sink/source config class is retrieved from the service definition (`sinkConfigClass` and `sourceConfigClass`)
- In case a field is stripped out, a WARN log is printed out:
```
11:31:37,446 WARN [public/default/elastic-sink-0] [instance: 0] JavaInstanceRunnable - Field 'bro' not defined in the SINK configuration org.apache.pulsar.io.elasticsearch.ElasticSearchConfig, the field will be ignored
```
- Added the configClass reference in the service definitions of connectors that were missing it

### Verifying this change

- [z] Make sure that the change passes the CI checks.

### Documentation

- [ ] `doc` 
- [ ] `doc-required` 
- [x] `doc-not-needed` 
- [ ] `doc-complete`

### Matching PR in forked repository

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][fn] Allow unknown fields in connectors config [pulsar]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.

codecov-commenter commented on PR #20116:
URL: https://github.com/apache/pulsar/pull/20116#issuecomment-4188137862

   ## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/20116?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :x: Patch coverage is `75.00000%` with `13 lines` in your changes missing coverage. Please review.
   :white_check_mark: Project coverage is 72.95%. Comparing base ([`4d8156f`](https://app.codecov.io/gh/apache/pulsar/commit/4d8156fca0582a350d30557d9243c851c4a45830?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`4cd1031`](https://app.codecov.io/gh/apache/pulsar/commit/4cd1031b13b93b03362404f7587afc7e3c90a58b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   :warning: Report is 2625 commits behind head on master.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/20116?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
   |---|---|---|
   | [...ulsar/functions/instance/JavaInstanceRunnable.java](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&filepath=pulsar-functions%2Finstance%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Ffunctions%2Finstance%2FJavaInstanceRunnable.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWZ1bmN0aW9ucy9pbnN0YW5jZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Z1bmN0aW9ucy9pbnN0YW5jZS9KYXZhSW5zdGFuY2VSdW5uYWJsZS5qYXZh) | 81.39% | [5 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   | [.../pulsar/functions/runtime/JavaInstanceStarter.java](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&filepath=pulsar-functions%2Fruntime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Ffunctions%2Fruntime%2FJavaInstanceStarter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWZ1bmN0aW9ucy9ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvZnVuY3Rpb25zL3J1bnRpbWUvSmF2YUluc3RhbmNlU3RhcnRlci5qYXZh) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   | [.../apache/pulsar/functions/runtime/RuntimeUtils.java](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&filepath=pulsar-functions%2Fruntime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Ffunctions%2Fruntime%2FRuntimeUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWZ1bmN0aW9ucy9ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvZnVuY3Rpb25zL3J1bnRpbWUvUnVudGltZVV0aWxzLmphdmE=) | 0.00% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   [![Impacted file tree graph](https://app.codecov.io/gh/apache/pulsar/pull/20116/graphs/tree.svg?width=650&height=150&src=pr&token=acYqCpsK9J&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   
   ```diff
   @@              Coverage Diff              @@
   ##             master   #20116       +/-   ##
   =============================================
   + Coverage     33.17%   72.95%   +39.77%     
   - Complexity    12225    31959    +19734     
   =============================================
     Files          1499     1868      +369     
     Lines        114308   138514    +24206     
     Branches      12414    15243     +2829     
   =============================================
   + Hits          37924   101050    +63126     
   + Misses        71445    29412    -42033     
   - Partials       4939     8052     +3113     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/pulsar/pull/20116/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [inttests](https://app.codecov.io/gh/apache/pulsar/pull/20116/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `24.16% <3.84%> (?)` | |
   | [systests](https://app.codecov.io/gh/apache/pulsar/pull/20116/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `24.80% <7.69%> (?)` | |
   | [unittests](https://app.codecov.io/gh/apache/pulsar/pull/20116/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.24% <75.00%> (+39.07%)` | :arrow_up: |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/20116?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [...ache/pulsar/functions/instance/InstanceConfig.java](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&filepath=pulsar-functions%2Finstance%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Ffunctions%2Finstance%2FInstanceConfig.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWZ1bmN0aW9ucy9pbnN0YW5jZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Z1bmN0aW9ucy9pbnN0YW5jZS9JbnN0YW5jZUNvbmZpZy5qYXZh) | `88.23% <100.00%> (+31.98%)` | :arrow_up: |
   | [...g/apache/pulsar/functions/worker/WorkerConfig.java](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&filepath=pulsar-functions%2Fruntime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Ffunctions%2Fworker%2FWorkerConfig.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWZ1bmN0aW9ucy9ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvZnVuY3Rpb25zL3dvcmtlci9Xb3JrZXJDb25maWcuamF2YQ==) | `92.85% <100.00%> (+58.17%)` | :arrow_up: |
   | [...ache/pulsar/functions/worker/FunctionActioner.java](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&filepath=pulsar-functions%2Fworker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Ffunctions%2Fworker%2FFunctionActioner.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWZ1bmN0aW9ucy93b3JrZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9mdW5jdGlvbnMvd29ya2VyL0Z1bmN0aW9uQWN0aW9uZXIuamF2YQ==) | `70.47% <100.00%> (+70.47%)` | :arrow_up: |
   | [.../apache/pulsar/functions/runtime/RuntimeUtils.java](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&filepath=pulsar-functions%2Fruntime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Ffunctions%2Fruntime%2FRuntimeUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWZ1bmN0aW9ucy9ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvZnVuY3Rpb25zL3J1bnRpbWUvUnVudGltZVV0aWxzLmphdmE=) | `77.96% <0.00%> (+73.26%)` | :arrow_up: |
   | [.../pulsar/functions/runtime/JavaInstanceStarter.java](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&filepath=pulsar-functions%2Fruntime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Ffunctions%2Fruntime%2FJavaInstanceStarter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWZ1bmN0aW9ucy9ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvZnVuY3Rpb25zL3J1bnRpbWUvSmF2YUluc3RhbmNlU3RhcnRlci5qYXZh) | `0.00% <0.00%> (ø)` | |
   | [...ulsar/functions/instance/JavaInstanceRunnable.java](https://app.codecov.io/gh/apache/pulsar/pull/20116?src=pr&el=tree&filepath=pulsar-functions%2Finstance%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Ffunctions%2Finstance%2FJavaInstanceRunnable.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWZ1bmN0aW9ucy9pbnN0YW5jZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Z1bmN0aW9ucy9pbnN0YW5jZS9KYXZhSW5zdGFuY2VSdW5uYWJsZS5qYXZh) | `70.46% <81.39%> (+21.27%)` | :arrow_up: |
   
   ... and [1539 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/20116/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   </details>
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] eolivelli commented on a diff in pull request #20116: [improve][fn] Allow unknown fields in connectors config

Posted by "eolivelli (via GitHub)" <gi...@apache.org>.

eolivelli commented on code in PR #20116:
URL: https://github.com/apache/pulsar/pull/20116#discussion_r1174842575


##########
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java:
##########
@@ -870,6 +872,56 @@ private void setupInput(ContextImpl contextImpl) throws Exception {
             Thread.currentThread().setContextClassLoader(this.instanceClassLoader);
         }
     }
+    private Map<String, Object> parseComponentConfig(String connectorConfigs) throws IOException {
+        return parseComponentConfig(connectorConfigs, instanceConfig, componentClassLoader, componentType);
+    }
+
+    static Map<String, Object> parseComponentConfig(String connectorConfigs,
+                                                    InstanceConfig instanceConfig,
+                                                    ClassLoader componentClassLoader,
+                                                    org.apache.pulsar.functions.proto.Function
+                                                            .FunctionDetails.ComponentType componentType)
+            throws IOException {
+        final Map<String, Object> config = ObjectMapperFactory
+                .getMapper()
+                .reader()
+                .forType(new TypeReference<Map<String, Object>>() {})
+                .readValue(connectorConfigs);
+        if (instanceConfig.isIgnoreUnknownConfigFields() && componentClassLoader instanceof NarClassLoader) {
+            final String configClass;
+            if (componentType == org.apache.pulsar.functions.proto.Function.FunctionDetails.ComponentType.SOURCE) {
+                configClass = ConnectorUtils
+                        .getConnectorDefinition((NarClassLoader) componentClassLoader).getSourceConfigClass();
+            } else if (componentType == org.apache.pulsar.functions.proto.Function.FunctionDetails.ComponentType.SINK) {
+                configClass =  ConnectorUtils
+                        .getConnectorDefinition((NarClassLoader) componentClassLoader).getSinkConfigClass();
+            } else {
+                return config;
+            }
+            if (configClass != null) {
+                final Object configInstance = Reflections.createInstance(configClass,
+                        Thread.currentThread().getContextClassLoader());
+                final List<String> allFields =

Review Comment:
   I am not sure that this is correct.
   
   ObjectMapper should follow Java Beans conventions and use getter/setters together with public fields.
   We should use some ObjectMapper utilities here in order to ensure that we are doing the right thing



##########
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java:
##########
@@ -870,6 +872,56 @@ private void setupInput(ContextImpl contextImpl) throws Exception {
             Thread.currentThread().setContextClassLoader(this.instanceClassLoader);
         }
     }
+    private Map<String, Object> parseComponentConfig(String connectorConfigs) throws IOException {
+        return parseComponentConfig(connectorConfigs, instanceConfig, componentClassLoader, componentType);
+    }
+
+    static Map<String, Object> parseComponentConfig(String connectorConfigs,
+                                                    InstanceConfig instanceConfig,
+                                                    ClassLoader componentClassLoader,
+                                                    org.apache.pulsar.functions.proto.Function
+                                                            .FunctionDetails.ComponentType componentType)
+            throws IOException {
+        final Map<String, Object> config = ObjectMapperFactory
+                .getMapper()
+                .reader()
+                .forType(new TypeReference<Map<String, Object>>() {})
+                .readValue(connectorConfigs);
+        if (instanceConfig.isIgnoreUnknownConfigFields() && componentClassLoader instanceof NarClassLoader) {
+            final String configClass;
+            if (componentType == org.apache.pulsar.functions.proto.Function.FunctionDetails.ComponentType.SOURCE) {
+                configClass = ConnectorUtils
+                        .getConnectorDefinition((NarClassLoader) componentClassLoader).getSourceConfigClass();
+            } else if (componentType == org.apache.pulsar.functions.proto.Function.FunctionDetails.ComponentType.SINK) {
+                configClass =  ConnectorUtils
+                        .getConnectorDefinition((NarClassLoader) componentClassLoader).getSinkConfigClass();
+            } else {
+                return config;
+            }
+            if (configClass != null) {
+                final Object configInstance = Reflections.createInstance(configClass,
+                        Thread.currentThread().getContextClassLoader());
+                final List<String> allFields =
+                        Reflections
+                                .getAllFields(configInstance.getClass())
+                                .stream()
+                                .map(Field::getName)
+                                .collect(Collectors.toList());
+
+                for (String s : config.keySet()) {
+                    if (!allFields.contains(s)) {
+                        log.warn("Field '{}' not defined in the {} configuration {}, the field will be ignored",

Review Comment:
   maybe this should be logged as "ERROR", WARNINGs tend to be ignored by alert systems



##########
pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java:
##########
@@ -738,6 +738,17 @@ public String getFunctionAuthProviderClassName() {
     )
     private List<String> additionalJavaRuntimeArguments = new ArrayList<>();
 
+    @FieldContext(
+            category = CATEGORY_CONNECTORS,
+            doc = "Whether to ignore unknown properties when deserializing the connector configuration. "
+                    + "After upgrading a connector to a new version with a new configuration, "
+                    + "the new configuration may not be compatible with the old connector. "
+                    + "In case of rollback, it's required to also rollback the connector configuration. "
+                    + "Ignoring unknown fields makes possible to keep the new configuration and "
+                    + "only rollback the connector."
+    )
+    private boolean ignoreUnknownConfigFields = false;

Review Comment:
   This configuration applies to the instances of the functions/connectors, and not to the function worker (that already  ignores unknown fields)
   
   what about 'functionsIgnoreUnknownConfigFields' ?    (maybe we can do better, but connectors are actually functions)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] Technoboy- closed pull request #20116: [improve][fn] Allow unknown fields in connectors config

Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.

Technoboy- closed pull request #20116: [improve][fn] Allow unknown fields in connectors config
URL: https://github.com/apache/pulsar/pull/20116


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org