Support eager evaluation of all Kafka filters #212

ankitk-me · 2023-04-19T16:14:50Z

jfallows · 2023-04-19T18:07:23Z

...-kafka.spec/src/main/java/io/aklivity/zilla/specs/binding/kafka/internal/KafkaFunctions.java

+                final KafkaMergedDataExFW mergedDataEx)
+            {
+                return filters == null || filters == mergedDataEx.filters();
+            }


Need to call matchFilters from match above.

jfallows · 2023-04-19T18:08:21Z

...ka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/internal/KafkaFunctionsTest.java

+                .header("name", "value")
+                .build()
+                .build()
+                .build();


Use further indentation to show nested .build() objects.

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

jfallows · 2023-04-20T18:14:36Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

@@ -962,6 +963,93 @@ public String toString()
            }
        }

+        private static final class EagerOr extends KafkaFilterCondition


As discussed, suggest creating a superclass Or and subsclasses LazyOr and EagerOr for improved readability.

Only the .test(...) method implementation will differ between LazyOr and EagerOr, so everything else can be implemented in the Or superclass.

jfallows · 2023-04-20T18:16:55Z

runtime/binding-kafka/src/main/zilla/internal.idl

@@ -36,6 +36,7 @@ scope internal
            kafka::KafkaHeader[] trailers;
            uint32 paddingLen;
            octets[paddingLen] padding;
+            int64 filters;


This change needs to be reverted as KafkaCacheEntry is shared across multiple readers, whereas filters mask applies to each individual cache client stream.

jfallows · 2023-04-20T18:18:17Z

...ka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/internal/KafkaFunctionsTest.java

@@ -440,6 +441,7 @@ public void shouldGenerateMergedDataExtension()
                                     .typeId(0x01)
                                     .merged()
                                         .timestamp(12345678L)
+                                         .filters(0L)


The example should probably show a non-zero filter mask, because filter mask zero means the message didn't match and therefore wouldn't be delivered.

jfallows · 2023-04-20T18:19:21Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

@@ -219,6 +226,7 @@ scope kafka
        {
            int32 deferred = 0;         // INIT only (TODO: move to DATA frame)
            int64 timestamp = 0;        // INIT only
+            int64 filters = 0;          // INIT only


Suggest we default to -1 instead, as filters mask 0 means no match.

jfallows · 2023-04-20T18:19:26Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

@@ -269,6 +278,7 @@ scope kafka
            int64 timestamp = 0;        // INIT only
            int32 headersSizeMax = 4;   // INIT only
            int64 producerId = -1;      // INIT only
+            int64 filters = 0;          // INIT only


Suggest we default to -1 instead, as filters mask 0 means no match.

jfallows · 2023-04-20T23:19:52Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

                KafkaCacheEntryFW cacheEntry)
            {
-                return cacheEntry != null;
+                return cacheEntry != null ? 1L : 0L;


Does this need to be -1 to generate the default, non-zero value for None filter?

jfallows · 2023-04-20T23:21:02Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

@@ -876,11 +881,11 @@ public String toString()
            }
        }

-        private static final class Or extends KafkaFilterCondition
+        private static class Or extends KafkaFilterCondition


Let's mark this abstract too.

jfallows · 2023-04-20T23:21:41Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

+            public long test(KafkaCacheEntryFW cacheEntry)
+            {
+                return 0L;
+            }


Remove method, now that class is abstract.

jfallows · 2023-04-20T23:21:59Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

+
+            @Override
+            public long test(
+                    KafkaCacheEntryFW cacheEntry)


Remove extra indentation.

jfallows · 2023-04-20T23:25:50Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

+                    if (result != 0L)
+                    {
+                        accept |= i + 1;
+                    }


Let's rework this code so that it can simplify to accept |= condition.test(cacheEntry);.

That requires the condition to return the relevant bitmask for the filter index it is on, not just 0 or 1.

Therefore, when each filter condition is created for non-Or conditions, it needs to know the filter index, so it can compute 1 << index as bitmask to return when .test(...) is successful, instead of 1.

jfallows · 2023-04-24T21:18:00Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

    }

    private static KafkaFilterCondition.Key initNullKeyInfo(
+        long mask,


Before adding the mask, we used this as a shared instance for all null key conditions, but now we need the mask to be correct per client, so the shared part should shift to the nullKeyRO and we now need to create the mask-specific Key instance.

jfallows · 2023-04-24T21:21:01Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

                {
                    nextEntry = null;
                }

+                filters =  nextEntry != null ? condition.test(nextEntry) : 0L;


Please call condition.test(nextEntry) just once.

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

jfallows · 2023-04-25T17:11:16Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

@@ -71,7 +73,7 @@ public KafkaCacheCursorFactory(
    {
        this.writeBuffer = writeBuffer;
        this.checksum = new CRC32C();
-        this.nullKeyInfo = initNullKeyInfo(checksum);
+        this.nullKeyRO = initNullKeyRO();


We were previously using this approach to avoid copying the null key value for each client filter.
I don't see where we are using nullKeyRO after the changes.

jfallows · 2023-04-25T17:31:44Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

@@ -71,7 +72,6 @@ public KafkaCacheCursorFactory(
    {
        this.writeBuffer = writeBuffer;
        this.checksum = new CRC32C();
-        this.nullKeyInfo = initNullKeyInfo(checksum);


I think you might have misinterpreted the feedback. 🙂

I was saying that we want to preserve the original optimization of not copying null keys all the time, just the approach is slightly different now because we cannot pre-create the entire filter condition.

jfallows · 2023-04-25T17:42:48Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCacheCursorFactory.java

+
+        return value == null ?
+            new KafkaFilterCondition.Key(mask, checksum, nullKeyRO) : new KafkaFilterCondition.Key(mask, checksum, key);
+


This is closer, but we still make the copy of nullKeyRO each time we construct a new KafkaFilterCondition.Key.

Follow the constructor hierarchy up through super to see the copy in Equals.

Kafka Functions Changes

ea97b25

jfallows requested changes Apr 19, 2023

View reviewed changes

Specs & Implementation changes

dd2583e

jfallows requested changes Apr 20, 2023

View reviewed changes

Implementation and addressing review comments

ba448ab

ankitk-me marked this pull request as ready for review April 20, 2023 20:53

jfallows requested changes Apr 20, 2023

View reviewed changes

ankitk-me added 3 commits April 21, 2023 09:56

Addressing review comments

960b913

Addressing Review Comments

e77aa55

minor refactoring

218aa78

jfallows requested changes Apr 24, 2023

View reviewed changes

Addressing review comments

d1da861

jfallows requested changes Apr 25, 2023

View reviewed changes

removing nullKeyRO

76a3d68

ankitk-me self-assigned this Apr 25, 2023

jfallows requested changes Apr 25, 2023

View reviewed changes

Addressing review comment

3a22735

jfallows requested changes Apr 25, 2023

View reviewed changes

Addressing review comment

a3e9b45

jfallows approved these changes Apr 26, 2023

View reviewed changes

jfallows merged commit b928955 into aklivity:develop Apr 26, 2023

ankitk-me deleted the eagerEvaluation branch May 26, 2023 10:11

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support eager evaluation of all Kafka filters #212

Support eager evaluation of all Kafka filters #212

ankitk-me commented Apr 19, 2023 •

edited by jfallows

Loading

jfallows Apr 19, 2023

jfallows Apr 19, 2023

jfallows Apr 20, 2023

jfallows Apr 20, 2023

jfallows Apr 20, 2023

jfallows Apr 20, 2023

jfallows Apr 20, 2023

jfallows Apr 20, 2023

jfallows Apr 20, 2023

jfallows Apr 20, 2023

jfallows Apr 20, 2023

jfallows Apr 20, 2023

jfallows Apr 24, 2023

jfallows Apr 24, 2023

jfallows Apr 25, 2023

jfallows Apr 25, 2023

jfallows Apr 25, 2023


		return value == null ?
		new KafkaFilterCondition.Key(mask, checksum, nullKeyRO) : new KafkaFilterCondition.Key(mask, checksum, key);

Support eager evaluation of all Kafka filters #212

Support eager evaluation of all Kafka filters #212

Conversation

ankitk-me commented Apr 19, 2023 • edited by jfallows 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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ankitk-me commented Apr 19, 2023 •

edited by jfallows

Loading