Use bytes for record keys instead of String #157

ajalab · 2022-06-07T08:07:52Z

Fixes #126.

Related: #61 (previous attempt to support non-String keys, but abandoned)

Motivation

In these days, decaton has been used to subscribe arbitrary topics. Such topics may not have String as keys and DecatonTask as value.

now Decaton is kind of "general consumer framework" as it supports to consume messages in arbitrary-format, we think we should make Decaton to support also non-String keys.
#126 (comment)

Currently, however, Decaton does not fully support topics with non-String keys. One of the major unsupported features is retry queuing as reported in #126.

This is due to the previous design decision that Decaton should only subscribe topics whose contents are produced by DecatonClient with String keys.

Supporting only ascii-printable String was a design-decision that made in early days of Decaton that all tasks are assumed to be produced by DecatonClient.
#126 (comment)

Solution

To unlock the whole features of Decaton for non-String keys, this PR updates the key type from String to byte[]. This change is originally proposed by @kawamuray in #61 (comment).

Please note that this PR contains breaking changes, which update signature of public interfaces and classes ~~like DecatonClient~~ EDIT Based on the comment, we decided to keep using String for DecatonClient. Only decaton-processor classes and underlying producers (e.g., DecatonTaskProducer) will be affected.

Basically, this PR is made of two changes below.

Add ArbitraryTopicTypeTest to reproduce the problem Retry fails if a non-ASCII string is used for the Kafka Record Key #126
Replace String key definition with byte[]. If the hash of a key is needed for Map or Set, wrap the key with ByteBuffer.

TODO

Consider how to print byte[] keys - toString won't be human-readable Use bytes for record keys instead of String #157 (review)
Consider injecting a serializer to DecatonClient, so that users can avoid serializing keys every time when they put a task Use bytes for record keys instead of String #157 (review): EDIT We try to keep using String key in DecatonClient as discussed in Use bytes for record keys instead of String #157 (review).
- ~~@ajalab will make a PoC~~

CLAassistant · 2022-06-07T08:08:01Z

All committers have signed the CLA.

mauhiz

Have you considered how to handle the places where the key is used in toString?
Would there be a way to inject the serializer / deserializer and making things generic instead of using only raw bytes?

mauhiz · 2022-06-07T09:03:36Z

...sor/src/main/java/com/linecorp/decaton/processor/runtime/internal/BlacklistedKeysFilter.java

+            return true;
+        }
+
+        final String stringKey = new String(key, StandardCharsets.UTF_8);


what if the key cannot be read a String? this can happen. https://stackoverflow.com/questions/70667113/why-cant-i-decode-any-byte-using-utf-8#comment124924879_70667113

it looks like ignoreKeys should keep a set of byte arrays instead

Here I assumed new String(key, StandardCharsets.UTF_8) never throws an exception. Instead, it silently accepts the malformed texts by converting some irregular values with a placeholder.

One commonly used approach in UTF-8 decoders is to replace any malformed UTF-8 sequence by a replacement character (U+FFFD), which looks a bit like an inverted question mark, or a similar symbol.
https://www.cl.cam.ac.uk/~mgk25/ucs/examples/UTF-8-test.txt

Also, I assumed that we support key denylist only when the target key is String,

Support key blacklisting only if the key is the type of String (the filter just tries to instantiate String from whatever the byte array is and do comparison which might always fail when key is not a type of String).
Make key deserializer of KafkaConsumer configurable #61 (comment)

If those assumptions are valid, I think we can just ignore non-UTF8 keys here because they will never match the configured keys stored in the denylist.

I see. It seems we might want to support other input for this denylist, maybe longs...
Let's leave the possibility open even if we don't implement it in this PR

ajalab · 2022-06-07T10:02:49Z

Have you considered how to handle the places where the key is used in toString?

Ah not yet. In this sense, this change will make it hard to debug it since printed keys in log or elsewhere won't be human-readable anymore.

I'm not sure what's the most preferred way to display it (print in hex or try interpreting as UTF-8). Do you guys have any ideas about that?

Would there be a way to inject the serializer / deserializer and making things generic instead of using only raw bytes?

One of the solutions I came up with is to inject a serializer into DecatonClientImpl, so that we can replace put(byte[] key, ...) with put(K key, ...) where K is a type variable for keys. I just avoided doing this since I found the following comment.

However changing key as generic type T consequence the mess that almost all classes of decaton to be a generic class which we would wanna avoid.
#61 (comment)

However, adding a type variable in DecatonClientImpl and its related interfaces/classes may not be a significant change. Let me try implementing it as another commit for comparison.

mauhiz · 2022-06-07T14:10:58Z

almost all classes of decaton to be a generic class which we would wanna avoid

I don't see how that would be a bad thing, TBH.

Jokes aside, maybe we need to dissociate the "context key" (String, human readable key) and "record key" (bytes from Kafka), and just give a way to convert between them (by default : a StringSerializer / StringDeserializer, but could be Long serdes + toString)

ocadaruma

Thanks for tackling this!

Overall strategy looks good though, here is my thoughts:

decaton-processor => +1 for changing ProcessingContext's key type to byte[] (which will be considered as breaking change).
- As @mauhiz mentioned, making thing generic would be an option though, it could introduce another type parameter in DecatonProcessor, so the impact for users will be huge.
- On the other hand, I suppose only limited users refer ProcessingContext#key in their code so only changing it to byte[] would be good compromise.
decaton-client => I think we can keep key type as String for DecatonClient.
- Because DecatonClient is just a default client to produce decaton tasks in "standard" format, rather than general producer to produce tasks in arbitrary key/value serialization format

WDYT? @kawamuray

kawamuray · 2022-06-10T11:22:17Z

@ocadaruma Agree with your comment.

I don't see how that would be a bad thing, TBH.

Before I made a significant refactoring for class hierarchies, decaton was holding task type as generic parameter almost at all classes in objects hierarchy, while the type parameter is referred only to hold a field for most of the classes, and it was really messy.
I then refactoring the decaton code to look like today's which encapsulate type parameter to the ProcessPipeline, which is created by users themselves and passed to ProcessorsBuilder, then eventually stored in ProcessUnit but without the type parameter (ProcessPipeline<?>), so now the structure represents the fact that decaton's core classes which doesn't rely on the task type stay agnostic for the type of the task.

It might be an option to add another type parameter for key, to the ProcessPIpeline and objects under that, but that will make us need to move the key filtering (BlacklistedKeysFilter) form ProcessorSubscription level to much later part, after queueing for subpartition, which makes the key filtering feature less-effective by letting these tasks to be filtered out once to get into the queue before they gets dropped, and I think it is a problem since the key filtering feature is commonly used to drop tasks from abusive user account (key = userid). This is what making it difficult (to choice) here, and I think fixing byte[] type as a key, while expecting string keys for enabling key filtering feature would be a good compromise.

I'm happy to hear if there's a better idea though.

ajalab · 2022-06-13T09:15:06Z

Thank you all for sharing your idea!

decaton-client => I think we can keep key type as String for DecatonClient.

I also agree for this strategy. Let me try to implement subsequent changes in this way.

ajalab · 2022-06-13T10:08:36Z

decaton-client => I think we can keep key type as String for DecatonClient.

Added 33d026e to let DecatonClient use String instead of byte[] as the previous implementation does.

Please note that the underlying producer (DecatonTaskProducer) still has to use byte[] keys. This is because it's used by DecatonTaskRetryQueuingProcessor, which has to produce a retry task based on a record in an arbitrary topic.

I believe that user-impact due to this change is still small, since DecatonTaskProducer is not expected to be used by applications.

ocadaruma

Left few feedbacks.
Besides, could you fix below points as well?

We need to fix here too: https://github.com/line/decaton/blob/master/processor/src/main/java/com/linecorp/decaton/processor/LoggingContext.java#L47
Let's mention that key-blocking feature expects only String key while Decaton accepts any type of key in CONFIG_IGNORE_KEYS's javadoc

ocadaruma · 2022-06-14T07:07:36Z

testing/src/main/java/com/linecorp/decaton/testing/processor/SerialProcessing.java

                                key -> new ArrayList<>()).add(record);
    }

    @Override
    public void doAssert() {
        // Checks there's no overlap between two consecutive records' processing time
-        for (Entry<String, List<ProcessedRecord>> entry : records.entrySet()) {
+        for (Entry<ByteBuffer, List<ProcessedRecord>> entry : records.entrySet()) {


Then we should convert key to string when generating assert message to get readable result here: https://github.com/line/decaton/pull/157/files#diff-141d497c7aafcef2217814fbe1d2ea75d00a4089212776327360030a935d2278L56

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/SubPartitioner.java

processor/src/it/java/com/linecorp/decaton/processor/ArbitraryTopicTypeTest.java

ocadaruma · 2022-06-14T07:22:27Z

processor/src/it/java/com/linecorp/decaton/processor/ArbitraryTopicTypeTest.java

+    }
+
+    @Test(timeout = 30000)
+    public void testPrintableAsciiStringKeyValue() throws Exception {


I guess String-key based scenario is already tested in other basic tests.
The intention to have another test here is?

I just thought that there were no test scenarios that produce String-key records without DecatonClient but an arbitrary Producer.

But it looks we already have such scenarios? Let me consider removing this one then

d79abd2 Replaced with byte key/value test case (or we don't need two tests?)

processor/src/it/java/com/linecorp/decaton/processor/ArbitraryTopicTypeTest.java

mauhiz · 2022-06-18T02:02:22Z

processor/src/it/java/com/linecorp/decaton/processor/CoreFunctionalityTest.java

@@ -153,17 +154,17 @@ public void testSingleThreadProcessing() throws Exception {
        // Note that this processing semantics is not be considered as Decaton specification which users can rely on.
        // Rather, this is just a expected behavior based on current implementation when we set concurrency to 1.
        ProcessingGuarantee noDuplicates = new ProcessingGuarantee() {
-            private final Map<String, List<TestTask>> produced = new HashMap<>();
-            private final Map<String, List<TestTask>> processed = new HashMap<>();
+            private final Map<ByteBuffer, List<TestTask>> produced = new HashMap<>();


ByteBuffer is mutable and its equals/hashcode methods depend on position/limit/remaining. Meaning, reading from it will affect findability in the map.

I don't know if is this overkill, but what about defining a new wrapper type, that
simply delegates equals/hashCode to Arrays.equals / Arrays.hashCode?

It could also cache hashCode (like String does)

And have a toString that just attempts to decode it with UTF-8

mauhiz · 2022-06-18T02:05:59Z

...sor/src/main/java/com/linecorp/decaton/processor/runtime/internal/BlacklistedKeysFilter.java

        // Preceding isEmpty() check is for reducing tiny overhead applied for each contains() by calling
        // Object#hashCode. Since ignoreKeys should be empty for most cases..
-        if (!ignoreKeys.isEmpty() && ignoreKeys.contains(record.key())) {


Reading this again, I thought we might want to changes ignoreKeys to a Set<ByteBuffer> to avoid the charset decoding step (=encode only once when reading blacklist config)

This is finally fixed by 6d3cc6c

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/TaskRequest.java

testing/src/main/java/com/linecorp/decaton/testing/processor/ProcessedRecord.java

For LoggingContext to print record keys in bytes as human-readable String, this commit introduces a new interface `KeyFormatter` that translates keys from bytes into `String`. This interface has a canonical implementation that reads the byte key as a UTF-8 byte sequence. For other places, this commit just prevents the byte keys from being printed, or prints it as String if it's inside a test case.

To avoid charset decoding step.

ajalab · 2022-06-20T08:48:54Z

Addressed all of the comments above. Notably, we introduced two new classes here.

EDIT (2022-06-21) Those were removed #157 (comment)

`TaskKey`

This is a tiny wrapper of record keys in byte[]. It equips overridden hashCode with cache, and is used to represent a set or map of record keys instead of ByteBuffer. (cf. #157 (comment))

`KeyFormatter`

This is a functional interface that translates keys from TaskKey into String. By default, a canonical converter that recognizes the byte key as UTF-8 char sequence is used. Users can inject a custom key formatter via ProcessorsBuilder to print out non-String keys (e.g., Long) in the log.

ajalab · 2022-06-20T08:51:57Z

Somehow I could only re-request a review of @ocadaruma but @mauhiz could you take a look too?

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/TaskKey.java

...sor/src/main/java/com/linecorp/decaton/processor/runtime/internal/BlacklistedKeysFilter.java

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/SubPartitioner.java

processor/src/main/java/com/linecorp/decaton/processor/formatter/KeyFormatter.java

This reverts commit fc63284.

This reverts commit 8c90fe5.

This reverts commit f991510.

ajalab · 2022-06-21T06:39:35Z

Based on the comments, I removed/replaced the following new classes.

TaskKey: Use bytes for record keys instead of String #157 (comment) We decided to use it only when hashCode computation is needed for HashMap or HashSet. Renamed it to HashableKey to emphasize the usage
KeyFormatter: Use bytes for record keys instead of String #157 (comment) We decided to print the byte[] keys as UTF-8 regardless of the actual type.

ajalab · 2022-06-21T09:11:56Z

Hmm, seems the CI failed because of unresolvable dependencies (due to CloudFlare incident?) and I have no right to rerun the workflow

* What went wrong:
A problem occurred configuring root project 'decaton'.
> Could not resolve all files for configuration ':classpath'.
   > Could not resolve me.champeau.jmh:jmh-gradle-plugin:0.6.6.
     Required by:
         project : > me.champeau.jmh:me.champeau.jmh.gradle.plugin:0.6.6
      > Could not resolve me.champeau.jmh:jmh-gradle-plugin:0.6.6.
         > Could not get resource 'https://plugins.gradle.org/m2/me/champeau/jmh/jmh-gradle-plugin/0.6.6/jmh-gradle-plugin-0.6.6.pom'.
            > Could not GET 'https://plugins-artifacts.gradle.org/me.champeau.jmh/jmh-gradle-plugin/0.6.6/562890d50aca0743c76181dfbc9fdc8a0750e8c47c978bfbb84472ea[41](https://github.com/line/decaton/runs/6979355735?check_suite_focus=true#step:4:42)1a0c5d/jmh-gradle-plugin-0.6.6.pom'. Received status code 500 from server: Internal Server Error
   > Could not resolve gradle.plugin.com.github.johnrengelman:shadow:7.1.1.
     Required by:
         project : > com.github.johnrengelman.shadow:com.github.johnrengelman.shadow.gradle.plugin:7.1.1
      > Skipped due to earlier error
   > Could not resolve io.franzbecker:gradle-lombok:5.0.0.
     Required by:
         project : > io.franzbecker.gradle-lombok:io.franzbecker.gradle-lombok.gradle.plugin:5.0.0
      > Skipped due to earlier error

ocadaruma

Left few trivial feedbacks but almost looks good!

processor/src/main/java/com/linecorp/decaton/processor/LoggingContext.java

testing/src/main/java/com/linecorp/decaton/testing/TestUtils.java

processor/src/main/java/com/linecorp/decaton/processor/HashableKey.java

mauhiz · 2022-06-22T01:41:32Z

processor/src/main/java/com/linecorp/decaton/processor/HashableKey.java

+    }
+
+    @Override
+    public String toString() {


can we delegate to a static, reusable method, e.g. for use in LoggingContext

I'm a bit afraid that adding a such static method like HashableByteArray.asString(byte[]) (note: ~~toString can't be used due to name collision~~ nope, we can use it) leads to give multiple responsibilities to the class.

It might be better to create a utility class with a static method ByteArrays.toString(byte[])? WDYT (cc. @ocadaruma)

The method may support both null array and non-null array.

I'm also not sure if we can make UTF-8 conversion as if it's a canonical way (ByteArrays.toString) to convert byte[] into String... Just keep using String(byte[], Charset) here and there will be an option?

Anyway let me add ByteArrays.toString as PoC

Added de1cb5f com.linecorp.decaton.processor.internal.ByteArrays with static toString method

- rename class and fields - make it `final` class - move to decaton.processor.runtime.internal

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/HashableByteArray.java

kawamuray

LGTM. Great progress, thanks!

ocadaruma

Left last comment

processor/src/main/java/com/linecorp/decaton/processor/internal/ByteArrays.java

ocadaruma

I found one more point to fix....
I hope this is the last. :)

processor/src/main/java/com/linecorp/decaton/processor/internal/HashableByteArray.java

ocadaruma

Thanks for the update.

LGTM!

ajalab added 2 commits June 7, 2022 17:12

Add ArbitraryTopicTypeTest

39ef2af

Use bytes for record keys instead of String

b3048ae

ajalab force-pushed the use-bytes-for-keys branch from 50e5c2c to b3048ae Compare June 7, 2022 08:12

mauhiz reviewed Jun 7, 2022

View reviewed changes

ocadaruma reviewed Jun 8, 2022

View reviewed changes

Use String for DecatonTask key in DecatonClient

33d026e

ajalab requested a review from ocadaruma June 14, 2022 05:50

ocadaruma reviewed Jun 14, 2022

View reviewed changes

ajalab added 4 commits June 17, 2022 12:27

Remove TODO comment about eliminating array copy

198033e

Add and use TestUtils.producer(String, Serializer, Serializer)

1dfb7a1

Replace redundant ArbitraryTopicTypeTest test case

d79abd2

Clarify only String keys are supported by CONFIG_IGNORE_KEYS

ae91f56

mauhiz reviewed Jun 18, 2022

View reviewed changes

ajalab added 4 commits June 20, 2022 16:26

Add final to KafkaClusterRule field in ArbitraryTopicTypeTest

ec75b24

Introduce TaskKey; a wrapper for keys in bytes

8c90fe5

Use TaskKey for BlacklistedKeysFilter

fc63284

To avoid charset decoding step.

ajalab requested a review from ocadaruma June 20, 2022 08:50

ocadaruma reviewed Jun 20, 2022

View reviewed changes

ajalab added 3 commits June 21, 2022 14:29

Revert "Use TaskKey for BlacklistedKeysFilter"

4f26f2e

This reverts commit fc63284.

Revert "Introduce TaskKey; a wrapper for keys in bytes"

3994ebc

This reverts commit 8c90fe5.

Revert "Display byte keys with a given key formatter"

f034907

This reverts commit f991510.

Add HashableKey to replace ByteBuffer usage for Map/Set

f69c303

ajalab added 3 commits June 21, 2022 15:43

Exclude byte[] key from toString target

cd07084

Use HashableKey for BlacklistedKeysFilter

6d3cc6c

Support null key in HashableKey

d2894c4

ajalab requested a review from ocadaruma June 21, 2022 09:10

ocadaruma reviewed Jun 21, 2022

View reviewed changes

mauhiz reviewed Jun 22, 2022

View reviewed changes

ajalab added 4 commits June 22, 2022 12:01

Print "null" instead of empty string if key is null

b9eac22

Make TestUtils.defaultProducerProps private again

bf3d42c

Redefine HashableKey as HashableByteArray

9d24b49

- rename class and fields - make it `final` class - move to decaton.processor.runtime.internal

Compute hashCode of bytes preemptively only once

0c6ee0a

ocadaruma reviewed Jun 22, 2022

View reviewed changes

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/HashableByteArray.java Outdated Show resolved Hide resolved

Move HashableByteArray to decaton.processor.internal

9d9db83

ajalab force-pushed the use-bytes-for-keys branch from f3a0b0b to 9d9db83 Compare June 22, 2022 09:00

Introduce ByteArrays.toString for reuse

a288fc6

ajalab force-pushed the use-bytes-for-keys branch from de1cb5f to a288fc6 Compare June 22, 2022 09:05

ajalab requested a review from ocadaruma June 22, 2022 09:11

kawamuray approved these changes Jun 22, 2022

View reviewed changes

ocadaruma reviewed Jun 22, 2022

View reviewed changes

processor/src/main/java/com/linecorp/decaton/processor/internal/ByteArrays.java Outdated Show resolved Hide resolved

Do not surround decoded array with quotes for consistency

6f69376

ajalab requested a review from ocadaruma June 23, 2022 02:58

ocadaruma reviewed Jun 24, 2022

View reviewed changes

processor/src/main/java/com/linecorp/decaton/processor/internal/HashableByteArray.java Outdated Show resolved Hide resolved

LINELINT-911 Remove unused HashableByteArray#getArray

46ce23c

ocadaruma approved these changes Jun 27, 2022

View reviewed changes

ocadaruma merged commit ae8dd85 into line:master Jun 27, 2022

ocadaruma added the breaking change Breaking change for a public API label Jun 27, 2022

ajalab deleted the use-bytes-for-keys branch June 28, 2022 04:13

ocadaruma mentioned this pull request Jul 26, 2022

Make key deserializer of KafkaConsumer configurable #61

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Use bytes for record keys instead of String #157

Use bytes for record keys instead of String #157

ajalab commented Jun 7, 2022 •

edited

Loading

CLAassistant commented Jun 7, 2022 •

edited

Loading

mauhiz left a comment

mauhiz Jun 7, 2022

ajalab Jun 7, 2022

mauhiz Jun 7, 2022

ajalab commented Jun 7, 2022 •

edited

Loading

mauhiz commented Jun 7, 2022

ocadaruma left a comment •

edited

Loading

kawamuray commented Jun 10, 2022 •

edited

Loading

ajalab commented Jun 13, 2022

ajalab commented Jun 13, 2022

ocadaruma left a comment

ocadaruma Jun 14, 2022

ocadaruma Jun 14, 2022

ajalab Jun 17, 2022

ajalab Jun 17, 2022

mauhiz Jun 18, 2022

mauhiz Jun 18, 2022

ajalab Jun 22, 2022

ajalab commented Jun 20, 2022 •

edited

Loading

ajalab commented Jun 20, 2022

ajalab commented Jun 21, 2022

ajalab commented Jun 21, 2022

ocadaruma left a comment

mauhiz Jun 22, 2022

ajalab Jun 22, 2022 •

edited

Loading

ajalab Jun 22, 2022

ajalab Jun 22, 2022

ajalab Jun 22, 2022

kawamuray left a comment

ocadaruma left a comment

ocadaruma left a comment

ocadaruma left a comment

Use bytes for record keys instead of String #157

Use bytes for record keys instead of String #157

Conversation

ajalab commented Jun 7, 2022 • edited Loading

Motivation

Solution

CLAassistant commented Jun 7, 2022 • edited Loading

mauhiz 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

ajalab commented Jun 7, 2022 • edited Loading

mauhiz commented Jun 7, 2022

ocadaruma left a comment • edited Loading

Choose a reason for hiding this comment

kawamuray commented Jun 10, 2022 • edited Loading

ajalab commented Jun 13, 2022

ajalab commented Jun 13, 2022

ocadaruma 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ajalab commented Jun 20, 2022 • edited Loading

TaskKey

KeyFormatter

ajalab commented Jun 20, 2022

ajalab commented Jun 21, 2022

ajalab commented Jun 21, 2022

ocadaruma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ajalab Jun 22, 2022 • 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

kawamuray left a comment

Choose a reason for hiding this comment

ocadaruma left a comment

Choose a reason for hiding this comment

ocadaruma left a comment

Choose a reason for hiding this comment

ocadaruma left a comment

Choose a reason for hiding this comment

ajalab commented Jun 7, 2022 •

edited

Loading

CLAassistant commented Jun 7, 2022 •

edited

Loading

ajalab commented Jun 7, 2022 •

edited

Loading

ocadaruma left a comment •

edited

Loading

kawamuray commented Jun 10, 2022 •

edited

Loading

ajalab commented Jun 20, 2022 •

edited

Loading

`TaskKey`

`KeyFormatter`

ajalab Jun 22, 2022 •

edited

Loading