KAFKA-13327, KAFKA-13328, KAFKA-13329: Clean up preflight connector validation by C0urante · Pull Request #11369 · apache/kafka

C0urante · 2021-09-29T14:02:32Z

KAFKA-13327: Modify preflight validation logic to prevent 500 responses from being returned instead of valid 200 responses with detailed error messages pertaining to the relevant configuration properties.

KAFKA-13328: Add preflight validation logic for per-connector header converters.

KAFKA-13329: Add preflight validation logic for per-connector key and value converter classes.

Additionally, a small bug in the logic for KAFKA-3829 introduced by KIP-458 is fixed; the preflight check to ensure that a sink connector's group ID doesn't conflict with the Connect worker's group ID now takes into account overrides made by the consumer.override.group.id connector property.

A new integration test is added that covers a wide variety of cases for sink connector, key/value converter, and header converter validation.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…alidation KAFKA-13327: Modify preflight validation logic to prevent 500 responses from being returned instead of valid 200 responses with detailed error messages pertaining to the relevant configuration properties. KAFKA-13328: Add preflight validation logic for per-connector header converters. KAFKA-13329: Add preflight validation logic for per-connector key and value converter classes.

C0urante · 2021-09-29T14:05:38Z

+        if (headerConverterClass == null
+            || headerConverterConfigValue == null
+            || !headerConverterConfigValue.errorMessages().isEmpty()
+        ) {
+            // Either no custom header converter was specified, or one was specified but there's a problem with it.
+            // No need to proceed any further.
+            return null;
+        }


The short-circuiting pattern here where we conditionally exit the method and otherwise proceed normally doesn't seem to be playing very nicely with Checkstyle's NPath complexity metric. I kept this style because it reduces indentation and seems more readable than the alternative (where the portions of code after an if statement such as this one would be put inside an else block), but it comes at the cost that we have to add a Checkstyle suppression.

C0urante · 2021-09-29T14:09:20Z

+                        // The value will be null if the class couldn't be found; no point in trying to load a ConfigDef for it
+                        if (value != null) {
+                            getConfigDefFromConfigProvidingClass(typeConfig, (Class<?>) value);
+                        }


This fixes an issue present in our unit tests where multiple error messages are added for a single invalid transform/predicate property when the class cannot be found. It's best if a single error message is returned stating that the class cannot be found; the second message (something along the lines of "invalid value null for property ...") did not add any value.

C0urante · 2021-09-29T14:10:46Z

            } catch (Exception e) {
-                throw new ConfigException(key, String.valueOf(cls), "Error getting config definition from " + baseClass.getSimpleName() + ": " + e.getMessage());
+                // Log the entire exception here in order to provide a stack trace that can be useful for debugging classloading issues
+                log.error("Failed to instantiate {} '{}'", baseClass.getSimpleName(), cls, e);


Added this error message based on discussion of a similar error-handling situation here.

C0urante · 2021-09-29T14:11:15Z

                        aliasKind + " is abstract and cannot be created. Did you mean " + childClassNames + "?";
                throw new ConfigException(key, String.valueOf(cls), message);
            }
-            T transformation;


Some basic cleanup to highlight the fact that this class is now agnostic with regards to plugin type and is not specific to SMTs.

C0urante · 2021-09-29T14:12:09Z

+        return ConnectorConfig.configDef()
+            .define(TOPICS_CONFIG, ConfigDef.Type.LIST, TOPICS_DEFAULT, ConfigDef.Importance.HIGH, TOPICS_DOC, COMMON_GROUP, 4, ConfigDef.Width.LONG, TOPICS_DISPLAY)
+            .define(TOPICS_REGEX_CONFIG, ConfigDef.Type.STRING, TOPICS_REGEX_DEFAULT, new RegexValidator(), ConfigDef.Importance.HIGH, TOPICS_REGEX_DOC, COMMON_GROUP, 4, ConfigDef.Width.LONG, TOPICS_REGEX_DISPLAY)
+            .define(DLQ_TOPIC_NAME_CONFIG, ConfigDef.Type.STRING, DLQ_TOPIC_DEFAULT, Importance.MEDIUM, DLQ_TOPIC_NAME_DOC, ERROR_GROUP, 6, ConfigDef.Width.MEDIUM, DLQ_TOPIC_DISPLAY)
+            .define(DLQ_TOPIC_REPLICATION_FACTOR_CONFIG, ConfigDef.Type.SHORT, DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DEFAULT, Importance.MEDIUM, DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DOC, ERROR_GROUP, 7, ConfigDef.Width.MEDIUM, DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DISPLAY)
+            .define(DLQ_CONTEXT_HEADERS_ENABLE_CONFIG, ConfigDef.Type.BOOLEAN, DLQ_CONTEXT_HEADERS_ENABLE_DEFAULT, Importance.MEDIUM, DLQ_CONTEXT_HEADERS_ENABLE_DOC, ERROR_GROUP, 8, ConfigDef.Width.MEDIUM, DLQ_CONTEXT_HEADERS_ENABLE_DISPLAY);


It's safer to use a new ConfigDef object every time in order to avoid issues like https://issues.apache.org/jira/browse/KAFKA-9950, which can be caused by reusing a ConfigDef that is then accidentally mutated by some downstream logic.

C0urante · 2021-09-29T14:13:07Z

+            throughput = Long.parseLong(props.getOrDefault("throughput", "-1"));
+            batchSize = Integer.parseInt(props.getOrDefault("messages.per.poll", "1"));


My IDE was nagging me about unnecessary boxing.

C0urante · 2021-09-29T14:14:18Z

            fail("Shouldn't be able to create connector");
        } catch (ConnectRestException e) {
-            assertEquals(e.statusCode(), 400);
+            assertEquals(400, e.statusCode());


The expected value should go first, as it's used in error messages that come up with test failures ("expected <first argument>, got <second argument>").

This test did fail once locally for me; I believe it was just some flakiness induced by my laptop being overworked.

gharris1727

It's great to see all of those validation test cases, and even better to see these validations actually making it all the way up to the user. Thanks so much @C0urante!

gharris1727 · 2021-09-30T21:16:24Z

+
+            Class<?> cls = (Class<?>) value;
+            try {
+                cls.getDeclaredConstructor().newInstance();


I hope these arbitrary classes that people put in here don't cause memory leaks from doing their own initialization.

Think it's worth adding a warning about that to the Javadoc for the class?

No, I don't think that buys us much, as the developer using this validator won't have much more insight into the arbitrary classes' behavior than the developer writing this validator.
Perhaps we could combine these two validators, or make the execution of this one dependent on the subclass validator, to try to narrow down the range of classes that could be instantiated here. For example, if the user is supposed to provide a HeaderConverter class, the typical HeaderConverter class will have a non-leaky default constructor.
...Except i just looked through the API definitions, and I don't think the javadocs for any of our API classes talk about default constructors and expectations, I'd have to dig deep to find out where this expectation is documented, if it exists. Maybe checking the subclass doesn't buy ourselves anything.

If you want to write a validator like this, I think the only justification we can have is that if it causes a memory leak, that's the implementation's fault because the framework never had any guarantees on the number of objects instantiated.

LGTM.

gharris1727 · 2021-09-30T21:44:22Z

        }
    }

+    private static class ConfigError {


This has the same contents as a plain ConfigException, but I think the reason you're using this class instead of the plain exception to avoid the cost of instantiating the exception and filling the stacktrace. I think this is a good consideration, but seeing as the existing ConfigDef infrastructure leverages the normal ConfigException in so many cases, I'm not sure this performance impact is substantial.

It's also desirable to return multiple errors from this one method, and that's not possible using an exception. I think passing in Map<String, ConfigValue> and calling the addErrorMessage function as a utility function directly avoids the need for this POJO, while getting you the ability to return multiple errors.

Also, Is there a reason that we can't have this validation return a ConfigInfos object, and merge it in late like the rest of the validations taking place in AbstractHerder::validateConnnectorConfig? That seems like the standard convention.

I think the reason you're using this class instead of the plain exception to avoid the cost of instantiating the exception and filling the stacktrace.

Not exactly; it's because the ConfigException class isn't really an effective POJO for a property, its value, and an error message. There are no getters for any of those fields, and two of the three constructors available would permit null values for some of them. I chose not to change the ConfigException class to add these getters since I'm not sure it's the best idea for an exception class to be doing double duty as a POJO as well.

I think passing in Map<String, ConfigValue> and calling the addErrorMessage function as a utility function directly avoids the need for this POJO, while getting you the ability to return multiple errors.

This is true, but I'm not sure how we could make that work cleanly with the other current use for SinkConnectorConfig::validate, which is to do a preliminary validation step during WorkerSinkTask initialization.

Also, Is there a reason that we can't have this validation return a ConfigInfos object, and merge it in late like the rest of the validations taking place in AbstractHerder::validateConnnectorConfig? That seems like the standard convention.

It'd make it a little more difficult for subclasses (i.e., DistributedHerder) to override behavior and add their own logic on top.

gharris1727 · 2021-09-30T22:03:55Z

+        Map<String, String> config = defaultSinkConnectorProps();
+        config.remove(TOPICS_CONFIG);
+        config.remove(TOPICS_REGEX_CONFIG);
+        connect.validateConnectorConfig(config.get(CONNECTOR_CLASS_CONFIG), config);


wait where are the assertions? isn't this supposed to fail?

Pitfalls of TDD. This was failing at the beginning as it is and after it started passing I forgot to go back and add a more fine-grained assertion.

Wow, good catch. Turns out all three of these tests were broken, including some genuine errors in the new validation logic.

gharris1727 · 2021-09-30T22:07:11Z

+    }
+
+    @Test
+    // TODO: Is this actually necessary? Should we permit the same SMT to be applied multiple times?


Was this a validation that already takes place, and you're just adding a test for it?
I think this makes sense because you can just work around it by renaming your transform name and copy-pasting the transform configurations.

Was this a validation that already takes place, and you're just adding a test for it?

Yep, exactly.

I think this makes sense because you can just work around it by renaming your transform name and copy-pasting the transform configurations.

I hear that, but at the same time I wonder if it's doing anybody any good to force them to copy+paste in the first place.

gharris1727 · 2021-09-30T22:08:48Z

    private static final Logger log = LoggerFactory.getLogger(EmbeddedConnectClusterAssertions.class);
    public static final long WORKER_SETUP_DURATION_MS = TimeUnit.SECONDS.toMillis(60);
-    public static final long VALIDATION_DURATION_MS = TimeUnit.SECONDS.toMillis(30);
+    public static final long VALIDATION_DURATION_MS = TimeUnit.SECONDS.toMillis(5);


Does this reduce test runtime? Might this introduce flakiness in a CI environment?

Might this introduce flakiness in a CI environment?

I highly doubt it; validations usually take place very quickly.

Does this reduce test runtime?

Sort of. I lowered the value because I used test-driven development for this PR and at the beginning some of my failing test cases were talking 30 seconds each. If we really want to play it safe I think 10 seconds should be fine. Or, alternatively, we could revamp the validation assertions to not use TestUtils::waitForCondition at all and only issue one validation request.

lowered the value because I used test-driven development for this PR and at the beginning some of my failing test cases were talking 30 seconds each.

Ah, this is a productivity win for everyone using the embedded assertions then. 👍

we could revamp the validation assertions to not use TestUtils::waitForCondition at all and only issue one validation request.

I don't think this is necessary, and any tests that rely on retries for validations would definitely flake out in CI.

…/AbstractHerder.java Co-authored-by: Greg Harris <gharris1727@gmail.com>

gharris1727

LGTM, Thanks @C0urante this is really going to help users get better feedback than a blind 500 error for these odd cases.

C0urante · 2023-06-06T17:18:44Z

Downgrading to a draft until I can revisit and fix the merge conflicts. I may also split this into several PRs to make review easier.

C0urante · 2023-08-28T15:59:17Z

A lot of merge conflicts have accrued on this one. Instead of resolving them all at once, I've decided to split this PR out into three smaller PRs, which should also make the review process easier.

The first of the three is #14303; others to come soon.

C0urante · 2023-08-29T19:55:41Z

#14304 and #14309 are the other two follow-ups.

C0urante commented Sep 29, 2021

View reviewed changes

Address Checkstyle grievances

7cd6b74

gharris1727 reviewed Sep 30, 2021

View reviewed changes

C0urante and others added 2 commits October 1, 2021 09:06

Update connect/runtime/src/main/java/org/apache/kafka/connect/runtime…

9f7a279

…/AbstractHerder.java Co-authored-by: Greg Harris <gharris1727@gmail.com>

Address review comments and fix sink connector config validation logic

03252c1

gharris1727 approved these changes Oct 1, 2021

View reviewed changes

kkonstantine added the connect label Oct 19, 2021

Update AbstractHerder.java

fa9531e

gharris1727 mentioned this pull request Jun 5, 2023

KAFKA-13756: Connect validate endpoint should return proper validatio… #13813

Closed

3 tasks

C0urante marked this pull request as draft June 6, 2023 17:18

C0urante mentioned this pull request Aug 28, 2023

KAFKA-13327: Gracefully report connector validation errors instead of returning 500 responses #14303

Merged

3 tasks

C0urante closed this Aug 28, 2023

C0urante deleted the kafka-13327-13328-13329 branch August 28, 2023 15:59

		throughput = Long.parseLong(props.getOrDefault("throughput", "-1"));
		batchSize = Integer.parseInt(props.getOrDefault("messages.per.poll", "1"));

Uh oh!

Conversation

C0urante commented Sep 29, 2021

Committer Checklist (excluded from commit message)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

C0urante Sep 29, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

C0urante Sep 29, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gharris1727 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

C0urante Oct 1, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

C0urante Oct 1, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

C0urante Oct 1, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gharris1727 left a comment

Choose a reason for hiding this comment

Uh oh!

C0urante commented Jun 6, 2023

Uh oh!

C0urante commented Aug 28, 2023

Uh oh!

C0urante commented Aug 29, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

C0urante Sep 29, 2021 •

edited

Loading

C0urante Sep 29, 2021 •

edited

Loading

C0urante Oct 1, 2021 •

edited

Loading

C0urante Oct 1, 2021 •

edited

Loading

C0urante Oct 1, 2021 •

edited

Loading