Add support for object change detection in filters #2

amajedi · 2018-11-13T21:23:44Z

~~NOTE: This PR is currently unstable and should not be merged. It suffers from issue #4.~~
Update: Issue #4 is not exclusive to this PR, removing [UNSTABLE] tag.

Why this PR is being opened:
This PR fixes issue #5 by allowing filters to detect changes. Given a One-To-Many relationship with frequent updates to a single record in the first relation, may cause an excessive number of output records to be produced if a large number of records corresponding to it exists in the second relation. This PR introduces the capability for filters to detect changes in input records and decide the course of action that should be taken, one of: UPDATE, DELETE, or SKIP in an effort to combat this problem.

Why existing functionality does not work:
The existing filter functionality only allows records to be treated as deleted (or tombstoned). This may cause output records to exclude necessary data.

What this PR changes:

BaseFilter requires a filter function which now returns FilterMode
BaseFilter.filter() additionally accepts the old BaseRecord as the last argument
Existing DefaultFilter class updated to be more explicit about previously implicit filtering logic
The next record is cached in a staging variable, this may have slight memory implications

Notes:

See Commit 5dbb83a Filtering happens in the KafkaTopicIterator.next() call. An important caveat is that hasNext() may return True if records exist in the input topic, however, if those records are all skipped, the return value of next() will be NULL. In an upcoming commit, I will be changing hasNext() to perform filtering as well to avoid this discrepancy
Commit 25f3212 updates BaseFilter to support two filter() signatures, one which accepts the old record and one which doesn't. This allows users who only care about the new record in their filters to avoid having an ignored argument.
Commit 415f0b0 corrects [known issue] Since KafkaTopic.hasNext() and KafkaTopic.next() can now theoretically consume many input topic messages as they skip along, they should also now increment records.consumed and records.consumed.by.entity. Previously, this was only done here. Without any changes, we will be under reporting these metrics.
Commit c0d719f addresses PR feedback related to large number of arguments necessary for BaseTopic configuration. Created a TopicConfig class, an instance can be passed to BaseTopic.configure instead.
Commit 168dcec addresses a bug which fails to null the value field on deletes, see: https://github.com/jwplayer/southpaw/blob/master/src/main/java/com/jwplayer/southpaw/topic/KafkaTopic.java#L91 this is used by the engine to avoid generating denorm docs
0b72cf1 removes deprecated function signature from BaseFilter. This was initially intended to be backwards compatible with existing filters but isn't due to change in return value.

TBoshoven · 2018-11-15T19:41:59Z

src/main/java/com/jwplayer/southpaw/filter/BaseFilter.java

@@ -25,6 +25,17 @@
 * in any denormalized records. They are effectively treated as a tombstone.
 */
 public abstract class BaseFilter {
+
+    /*


Docstrings should start with /**.
There are some other cases of missing * in this PR.

TBoshoven · 2018-11-15T19:56:00Z

src/main/java/com/jwplayer/southpaw/filter/BaseFilter.java

     */
-    public abstract boolean isFiltered(String entity, BaseRecord record);
+    public FilterMode filter(String entity, BaseRecord record, BaseRecord oldRecord) {


Might be good to keep that last parameter optional.

It is optional in the sense that a class extending the BaseFilter has the option of implementing either function.

TBoshoven · 2018-11-15T20:18:27Z

src/main/java/com/jwplayer/southpaw/topic/KafkaTopic.java

        protected Iterator<ConsumerRecord<byte[], byte[]>> iter;
        protected KafkaTopic<K, V> topic;

+        // Information about the next valid record


Might be good to describe what null values mean.

TBoshoven · 2018-11-15T20:20:32Z

src/main/java/com/jwplayer/southpaw/topic/KafkaTopic.java

+        /*
+         * Internal helper to obtain and stage the next non-skipped record
+         */
+        private ConsumerRecord<byte[], byte[]> getAndStageNextRecord() {


getAndStage is a bit confusing to me. Maybe findAndStage or something, implying that this is more than a simple getter.

TBoshoven · 2018-11-15T20:23:10Z

src/main/java/com/jwplayer/southpaw/topic/KafkaTopic.java

        }

        @Override
        public boolean hasNext() {
-            return iter.hasNext();
+            return (getAndStageNextRecord() != null);


() are superfluous

TBoshoven · 2018-11-15T20:24:34Z

src/main/java/com/jwplayer/southpaw/topic/KafkaTopic.java

@@ -69,49 +76,125 @@
        private KafkaTopicIterator(Iterator<ConsumerRecord<byte[], byte[]>> iter, KafkaTopic<K, V> topic) {
            this.iter = iter;
            this.topic = topic;
+            this.nextRecord = null;


this.resetStagedRecord() ?

src/main/java/com/jwplayer/southpaw/filter/BaseFilter.java

src/main/java/com/jwplayer/southpaw/topic/BaseTopic.java

ksindi

Looks great!

TBoshoven

Went over the changes, LGTM.
Naturally, I could not approve without a comment about a newline ;)

TBoshoven · 2018-11-28T09:35:10Z

src/main/java/com/jwplayer/southpaw/filter/BaseFilter.java

@@ -34,10 +45,26 @@ public void configure(Map<String, Object> config) {
    }

    /**
-     * Determines if the given record should be filtered based on its entity
+     * DEPRECATED


Use this instead.

TBoshoven · 2018-11-28T09:36:17Z

src/main/java/com/jwplayer/southpaw/metric/StaticGauge.java

+    public void update(T value) {
+        this.value = value;
+    }
+}


Missing newline, looks like your editor needs to be configured.

…sage

…cted now

…onstructor

eric-weaver · 2019-01-08T16:24:31Z

src/main/java/com/jwplayer/southpaw/filter/BaseFilter.java

     */
-    public abstract boolean isFiltered(String entity, BaseRecord record);
+    public FilterMode filter(String entity, BaseRecord record) {


Continuing from @moritastic comment, this signature isn't deprecated. It's a breaking change since the return value changes. This PR already introduces a number of backwards compatibility changes. Lets just remove this and I'll make a note of it in the release

I agree, 0b72cf1 removes the deprecated signature

…ible change

amajedi requested review from moritastic and eric-weaver November 13, 2018 21:40

amajedi changed the title ~~[WIP] Support for change detection in filters~~ Support for change detection in filters Nov 14, 2018

TBoshoven reviewed Nov 15, 2018

View reviewed changes

moritastic requested changes Nov 19, 2018

View reviewed changes

src/main/java/com/jwplayer/southpaw/filter/BaseFilter.java Outdated Show resolved Hide resolved

src/main/java/com/jwplayer/southpaw/topic/BaseTopic.java Outdated Show resolved Hide resolved

moritastic approved these changes Nov 19, 2018

View reviewed changes

ksindi approved these changes Nov 21, 2018

View reviewed changes

TBoshoven approved these changes Nov 28, 2018

View reviewed changes

amajedi changed the title ~~Support for change detection in filters~~ [UNSTABLE] Support for change detection in filters Nov 30, 2018

amajedi changed the title ~~[UNSTABLE] Support for change detection in filters~~ Support for change detection in filters Dec 14, 2018

amajedi changed the title ~~Support for change detection in filters~~ Fix for issue #5: Add support for object change detection in filters Jan 2, 2019

amajedi changed the title ~~Fix for issue #5: Add support for object change detection in filters~~ Fix for issue #4: Add support for object change detection in filters Jan 2, 2019

amajedi changed the title ~~Fix for issue #4: Add support for object change detection in filters~~ Add support for object change detection in filters Jan 2, 2019

amajedi added 14 commits January 2, 2019 11:50

Support for change detection in filters

45aea73

Fix TestFilter so that e2e tests pass

915e79f

e2e test to ensure skipped messages do not appear in output topic mes…

7f57819

…sage

the hasNext() func now takes into account filtering

1c564bd

can remove null-check after next() call since hasNext() works as expe…

4f76c1c

…cted now

More flexible filter signatures

e6ce935

Documentation

0264d60

Separate Metrics file; fix under reporting issue

fd7ba3d

comments

035543f

Introduce TopicConfig to address large num of params in BaseTopic's c…

15283a6

…onstructor

filter signature deprecated

f16023b

Cleaning up unused imports

02038aa

error logging on failed delete

b085234

Return NULL value on delete so caller knows to avoid generating docs

a0678a3

amajedi force-pushed the change-detection-filtering branch from 168dcec to a0678a3 Compare January 2, 2019 16:51

eric-weaver reviewed Jan 8, 2019

View reviewed changes

Removing deprecated filter signature since it is a backwards incompat…

0b72cf1

…ible change

eric-weaver approved these changes Jan 8, 2019

View reviewed changes

eric-weaver merged commit 6cdd913 into master Jan 8, 2019

eric-weaver deleted the change-detection-filtering branch January 8, 2019 18:13

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add support for object change detection in filters #2

Add support for object change detection in filters #2

amajedi commented Nov 13, 2018 •

edited

Loading

TBoshoven Nov 15, 2018

TBoshoven Nov 15, 2018

amajedi Nov 15, 2018

TBoshoven Nov 15, 2018

TBoshoven Nov 15, 2018

TBoshoven Nov 15, 2018

TBoshoven Nov 15, 2018

ksindi left a comment

TBoshoven left a comment

TBoshoven Nov 28, 2018

TBoshoven Nov 28, 2018

eric-weaver Jan 8, 2019

amajedi Jan 8, 2019

Add support for object change detection in filters #2

Add support for object change detection in filters #2

Conversation

amajedi commented Nov 13, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ksindi left a comment

Choose a reason for hiding this comment

TBoshoven left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

amajedi commented Nov 13, 2018 •

edited

Loading