KAFKA-9673: Filter and Conditional SMTs by tombentley · Pull Request #8699 · apache/kafka

tombentley · 2020-05-20T18:33:59Z

Add Predicate interface
Add Filter SMT
Add the predicate implementations defined in the KIP.
Create abstraction in ConnectorConfig for configuring Transformations and Connectors with the "alias prefix" mechanism
Add tests and fix existing tests.

Committer Checklist (excluded from commit message)

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

tombentley · 2020-05-20T18:36:33Z

Still need to review test coverage, but @kkonstantine, @mimaison, @bbejeck you might want to give it an initial pass.

tombentley · 2020-05-20T19:37:25Z

@C0urante you might also want to take a look.

C0urante

Looking pretty good @tombentley! Thanks for the ping.

I was able to get through all of the non-test files except for ConnectorConfig; will try to return and make another pass by the end of the week to cover the rest.

Also, do you think it might be worth it to write some integration tests for this stuff? I'm thinking at the very least it'd be cool to verify that Predicates work end-to-end for both source and sink connectors, and since we'd need to pick some SMTs to test the functionality of the Predicate plugin with, it might be a convenient excuse to also end-to-end test the new Filter SMT introduced here.

C0urante · 2020-05-20T20:51:08Z

Same comment here as with Filter; probably want to return a non-null ConfigDef here.

This is related to the issue discussed about configure().

We could return an empty ConfigDef here, but that would be a lie which could ultimately lead some other error if someone tried to use it with configure().

We can't invent a ConfigDef schema for this because the PredicatedTransformer would need to know about the Transformer is was going to be wrapping, but it can't know that before it's been configured with at least the Transformer's ConfigDef and it can't be configured before config() has been called. So we have a chicken and egg problem. Something (ConnectorConfig) must have some a priori knowledge of either PredicatedTransformer's ConfigDef, or know how to configure it without needing to call config() at all.

Since PredicatedTransformer is a purely internal class which will never be directly exposed to Connect users, we're not obliged to stick to the contract of config() and configure(). i.e. So both PredicatedTransformer.config and PredicatedTransformer.configure can throw when called, since we know no one else can call them and we know ConnectorConfig never will.

Hmmm... I think there might be some awkwardness here with trying to make PredicatedTransformer implement the Transformation interface. Could we replace every Transformation in the TransformationChain's transformation list with a PredicatedTransformer and, if there are no predicates configured for a transform by the user, make the default behavior for the PredicatedTransformer class to blindly apply its transformation?

This would solve a few problems:

No risk of users trying to actually use a PredicatedTransformer in a connector config, which they may try to do if we don't add logic to prevent it from being picked up during plugin path scanning on startup and logged as an SMT plugin

No need to implement methods that aren't used

One code path instead of two for application of transformations

More flexibility in instantiation and, possibly, the ability to encapsulate some of the ConfigDef generation logic in a separate class from ConnectorConfig (haven't looked into the specifics of this yet so may not actually be feasible or that elegant)

C0urante · 2020-05-20T21:46:56Z

Do we need to remove these properties here, or can we just read them? Removing might cause issues with SMTs that have config properties with these names; would leaving them in be likely to cause issues as well?

The compatibility section of the KIP says that if a connector already has these configs then they'll be masked by the new implicit configs. If we don't remove them here then we'd be passing the KIP-585 configs to a connector which had it's own semantics for those config keys, which would be incorrect.

Hmmm... wish we'd caught that earlier. Seems safer to just leave the properties in, but unless we want to call for a re-vote and an extension on the KIP deadline guess we'll have to keep this as-is.

tombentley · 2020-05-22T07:50:57Z

@C0urante thanks for the review, some excellent points there! I think an integration test is a great idea, which I'll work on next. I've addressed all your other comments.

kkonstantine

Thanks for the PR @tombentley. This looks in the right direction.
I managed to do a first pass. Comments are mostly minor.

Yet, we'll need to consider correct that the new package and interfaces are correctly included or excluded in PluginUtils to make these classes compatible with classloading isolation.

* Add Predicate interface * Add Filter SMT * Add the predicate implementations defined in the KIP. * Create abstraction in ConnectorConfig for configuring Transformations and Connectors with the "alias prefix" mechanism * Add tests and fix existing tests.

tombentley · 2020-05-27T10:58:56Z

Rebased for conflict.

@kkonstantine I've addressed those first comments, thanks! Still some work on the integration test (not passing when run via gradle).

rhauch · 2020-05-27T16:04:49Z

ok to test

rhauch

Thanks, @tombentley. I made a first pass, and for the most part this looks good. A few comments/questions below. I'll keep reviewing in more detail.

rhauch · 2020-05-27T16:33:26Z

+                .build();
+
+        // start the clusters
+        connect.start();


Should we wait until all brokers and Connect workers are available, via something like:

connect.assertions().assertExactlyNumBrokersAreUp(numBrokers, "Brokers did not start in time."); connect.assertions().assertExactlyNumWorkersAreUp(numWorkers, "Worker did not start in time.");

rhauch · 2020-05-27T16:35:32Z

+        connectorHandle.expectedCommits(numFooRecords);
+
+        // start a sink connector
+        connect.configureConnector(CONNECTOR_NAME, props);


This is an asynchronous method, and it's likely the connector will not be started and running before the test proceeds to the next statements. This can lead to very flaky tests.

We could instead wait until the connector is actually running, using something like:

connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS, "Connector tasks did not start in time.");

rhauch · 2020-05-27T16:43:59Z

+        String value = simpleConfig.getString(PATTERN_CONFIG);
+        try {
+            result = Pattern.compile(value);
+        } catch (PatternSyntaxException e) {
+            throw new ConfigException(PATTERN_CONFIG, value, "entry must be a Java-compatible regular expression: " + e.getMessage());
+        }


Can we ever get to line 64? The constructor of the config (line 58) should fail if the pattern validator fails to ensure the pattern is a valid regex, which means that if we make it past 58 then line 62 will never fail.

Am I missing something?

Co-authored-by: Randall Hauch <rhauch@gmail.com>

rhauch

Got through the test code, and found a few more suggestions.

rhauch · 2020-05-27T18:00:39Z

+        }
+    }
+
+    @Ignore("Is this really an error. There's no actual need for the predicates config (unlike transforms where it defines the order).")


It'd be good to have the test reflect the current behavior.

rhauch · 2020-05-27T18:02:56Z

+        try {
+            new ConnectorConfig(MOCK_PLUGINS, props);
+            fail();
+        } catch (ConfigException e) {
+            assertTrue(e.getMessage().contains("Value must be at least 80"));
+        }


We've moved to using assertThrows here, which would look something like:

Suggested change

try {

new ConnectorConfig(MOCK_PLUGINS, props);

fail();

} catch (ConfigException e) {

assertTrue(e.getMessage().contains("Value must be at least 80"));

}

ConfigException e = assertThrows(ConfigException.class, () -> new ConnectorConfig(MOCK_PLUGINS, props));

assertTrue(e.getMessage().contains("Value must be at least 42"));

rhauch · 2020-05-27T18:11:10Z

ok to test

…d more validation to TopicNameMatches

rhauch · 2020-05-28T00:00:26Z

Added several commits that fixed the failing unit and integration tests, and addressed several of my other comments.

rhauch · 2020-05-28T00:00:32Z

ok to test

rhauch · 2020-05-28T00:00:45Z

retest this please

rhauch · 2020-05-28T01:21:21Z

retest this please

rhauch · 2020-05-28T03:20:06Z

Builds were all aborted: Aborted by user anonymous

rhauch · 2020-05-28T03:20:10Z

retest this please

rhauch

Thanks, @tombentley! LGTM, pending a green build (which are currently waiting for executors).

rhauch · 2020-05-28T03:38:42Z

ok to test

rhauch · 2020-05-28T03:53:41Z

Jenkins jobs were never associated with this PR (see INFRA-20344), but here are the jobs for the latest commit:

kkonstantine · 2020-05-28T06:51:26Z

retest this please

kkonstantine

LGTM, pending a green build
Nice work @tombentley !

tombentley · 2020-05-28T08:26:14Z

@rhauch @kkonstantine thanks very much for making those fixes for me, I appreciate the time that must've taken. I've made one more trivial correction to the integration test.

rhauch · 2020-05-28T13:52:07Z

Thanks, @tombentley. Merging with 3 green builds that Jenkins did not tie back to this PR.

* 'trunk' of github.com:apache/kafka: (36 commits) Remove redundant `containsKey` call in KafkaProducer (apache#8761) KAFKA-9494; Include additional metadata information in DescribeConfig response (KIP-569) (apache#8723) KAFKA-10061; Fix flaky `ReassignPartitionsIntegrationTest.testCancellation` (apache#8749) KAFKA-9130; KIP-518 Allow listing consumer groups per state (apache#8238) KAFKA-9501: convert between active and standby without closing stores (apache#8248) KAFKA-10056; Ensure consumer metadata contains new topics on subscription change (apache#8739) MINOR: Log the reason for coordinator discovery failure (apache#8747) KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException (apache#8705) MINOR: remove unnecessary timeout for admin request (apache#8738) MINOR: Relax Percentiles test (apache#8748) MINOR: regression test for task assignor config (apache#8743) MINOR: Update documentation.html to refer to 2.6 (apache#8745) MINOR: Update documentation.html to refer to 2.5 (apache#8744) KAFKA-9673: Filter and Conditional SMTs (apache#8699) KAFKA-9971: Error Reporting in Sink Connectors (KIP-610) (apache#8720) KAFKA-10052: Harden assertion of topic settings in Connect integration tests (apache#8735) MINOR: Slight MetadataCache tweaks to avoid unnecessary work (apache#8728) KAFKA-9802; Increase transaction timeout in system tests to reduce flakiness (apache#8736) KAFKA-10050: kafka_log4j_appender.py fixed for JDK11 (apache#8731) KAFKA-9146: Add option to force delete active members in StreamsResetter (apache#8589) ... # Conflicts: # core/src/main/scala/kafka/log/Log.scala

C0urante reviewed May 20, 2020

View reviewed changes

kkonstantine added the connect label May 21, 2020

kkonstantine reviewed May 27, 2020

View reviewed changes

tombentley added 9 commits May 27, 2020 11:47

Use Predicates config group

2cfe4e1

Some review comments

0f0e138

fixup

6477fe9

Comment

460c5f8

tidy and fix a couple of tests

fba43c8

Numerous fixes and an integration test.

f55e146

Review comments

49b362c

Review comments

e76a0b4

tombentley force-pushed the KAFKA-9673-conditional-smt branch from 9d0ec29 to e76a0b4 Compare May 27, 2020 10:57

rhauch reviewed May 27, 2020

View reviewed changes

Apply suggestions from code review

602a89d

Co-authored-by: Randall Hauch <rhauch@gmail.com>

rhauch reviewed May 27, 2020

View reviewed changes

rhauch added 2 commits May 27, 2020 17:39

KAFKA-9673: Fix a few unit and integration tests

03ce156

KAFKA-9673: Added some tests, made others a bit more robust, and adde…

60cd39e

…d more validation to TopicNameMatches

rhauch approved these changes May 28, 2020

View reviewed changes

kkonstantine approved these changes May 28, 2020

View reviewed changes

Trivial correction

cbc8981

rhauch merged commit 1c4eb1a into apache:trunk May 28, 2020

Conversation

tombentley commented May 20, 2020

Committer Checklist (excluded from commit message)

Uh oh!

tombentley commented May 20, 2020

Uh oh!

tombentley commented May 20, 2020

Uh oh!

C0urante left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

tombentley commented May 22, 2020

Uh oh!

kkonstantine left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

tombentley commented May 27, 2020

Uh oh!

rhauch commented May 27, 2020

Uh oh!

rhauch left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rhauch left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rhauch commented May 27, 2020

Uh oh!

rhauch commented May 28, 2020

Uh oh!

rhauch commented May 28, 2020

Uh oh!

rhauch commented May 28, 2020

Uh oh!

rhauch commented May 28, 2020

rhauch commented May 28, 2020 •

edited

Loading