KAFKA-4714: Flatten and Cast single message transforms (KIP-66) #2458

ewencp · 2017-01-27T07:27:04Z

No description provided.

ewencp · 2017-01-27T07:30:58Z

@kkonstantine @hachikuji Could you review? These are 2 of the last 3 built-in SMTs we have from KIP-66. I'd like to get these in since we've seen asks for both on the mailing lists (directly, for casting when the connector can't constrain the range of values, and indirectly for Flatten for use with connectors that can only handle flat schemas).

The final SMT missing is TimestampConverter, which I think is really useful, but probably too complex (both implementation + testing) to squeeze in. However, these two are useful enough that I think we should push them through if possible.

asfbot · 2017-01-27T09:41:00Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1285/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-27T09:41:12Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1283/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-27T11:50:44Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1283/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-27T18:05:14Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1294/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-27T18:15:12Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1296/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-27T18:45:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1294/
Test PASSed (JDK 8 and Scala 2.12).

kkonstantine

Just a few comments. Looks ok to me.

kkonstantine · 2017-01-27T22:32:02Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+
+    @Override
+    public R apply(R record) {
+        if (operatingSchema(record) == null) {


Could have been a wonderful one-liner, but now I see that everywhere this version of apply is the same.

Not the same everywhere, but in many places yes. Can always clean up and simplify later, if it's worth it. I also mentioned in a previous PR to @shikhar that there is a fair amount of duplicated code that might be removed if we provided a common base class for transformations that fit a few requirements (e.g., support schema & schemaless versions requiring struct/map, can be applied to either Key or Value, etc).

I agree to punt.

kkonstantine · 2017-01-27T22:39:54Z

connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java

+public class CastTest {
+
+    @Test
+    public void castWholeRecordKeySchemad() {


Is Schemad a typo everywhere in this test? Or is it meant as an adjective indeed?

Meant as an adjective. Could easily swap for WithSchema. I debated on this and was unhappy with all the options I came up with :/

I'm up for creating new words.

I like withSchema. I also wondered if it was a typo and others probably will too.

changed to WithSchema

kkonstantine · 2017-01-27T22:54:48Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+
+    // As a special case for casting the entire value (e.g. the incoming key is a int64 but you know it could be an
+    // int32 and want the smaller width), we use an otherwise invalid field name in the cast spec to track this.
+    private static final String WHOLE_VALUE_CAST = "";


Does this refer to "identity conversion"? Is this a better name?

No, it's not identity. The original idea for the transformation was casting the type of fields (in a Struct w/ schemas, in a Map w/o schemas). But then I realized that if you apply it to keys, it might need to apply to the entire record (either key or value in the ConnectRecord) rather than a field. I was definitely stretching to find a name to refer to this...

Relatedly, there's some future improvements to be added here. Currently the code only handles casting the entire value or fields within the value if it is a Struct/Map. Still missing is handling hierarchical structures, i.e. in the spec, a field would ideally support some sort of dotted notation to allow for casting in nested structs.

I see now. But that's a case where an example has an edge in intuition.

kkonstantine · 2017-01-27T23:17:44Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+                        throw new ConnectException("Unexpected type in Cast transformation: " + value.getClass());
+                case BOOLEAN:
+                    if (value instanceof Number)
+                        return ((Number) value).longValue() > 0L;


Should this return false only on 0? Instead of <=0?

Ack. Fixed.

Seems like we still return false on negative numbers. Did the fix get included?

kkonstantine · 2017-01-27T23:23:45Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+    private static Map<String, Schema.Type> parseFieldTypes(List<String> mappings) {
+        final Map<String, Schema.Type> m = new HashMap<>();
+        for (String mapping : mappings) {
+            final String[] parts = mapping.split(":");


Not trimming the parts after split, feels a bit restrictive. Do we want to use a util method that does both?
Change shouldn't have big impact because I see it's only used here and in ReplaceField transform.

ConfigDef already handles trimming, so we shouldn't need to worry about it here.

I was referring to the parts after split though:

int8 : int32 vs
int8:int32

ack, added trim() calls before using the values

kkonstantine · 2017-01-27T23:36:11Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.kafka.connect.transforms.util.Requirements.requireMap;


Can't leave a comment in Requirements, but requireMap needs an annotation to suppress unchecked cast exception to avoid a warning.

What command complains?

I just noticed by reading and then ran Intellij's code inspection.
I see -Xlint:unchecked is not enabled in our gradle script, so we won't see it for now.
https://github.com/apache/kafka/blob/trunk/build.gradle#L128

Up to you.

kkonstantine · 2017-01-27T23:40:40Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Flatten.java

+                    throw new ConnectException("Flatten transformation does not support " + entry.getValue().getClass()
+                            + " for record without schemas (for field " + fieldName + ").");
+            }
+            entry.getValue();


Seems like a no-op

Good catch, was just some random leftover code.

kkonstantine · 2017-01-27T23:52:03Z

connect/transforms/src/test/java/org/apache/kafka/connect/transforms/FlattenTest.java

+    public void topLevelStructRequired() {
+        final Flatten<SourceRecord> xform = new Flatten.Value<>();
+        xform.configure(Collections.<String, String>emptyMap());
+        xform.apply(new SourceRecord(null, null,


Seems to fit in one line

Yeah, this was copy/paste. It is somewhat cleaner in that the three lines break up source partition/offset, topic/partition, and value schema/value. But I'm not tied to that split, so I've realigned them all on the same line.

I thought so, but wasn't excited with the grouping and it's not a long line anyways.

kkonstantine · 2017-01-27T23:52:15Z

connect/transforms/src/test/java/org/apache/kafka/connect/transforms/FlattenTest.java

+    public void topLevelMapRequired() {
+        final Flatten<SourceRecord> xform = new Flatten.Value<>();
+        xform.configure(Collections.<String, String>emptyMap());
+        xform.apply(new SourceRecord(null, null,


Just one line?

Same here as well.

kkonstantine · 2017-01-28T00:01:18Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Flatten.java

+    private void applySchemaless(Map<String, Object> originalRecord, String fieldNamePrefix, Map<String, Object> newRecord) {
+        for (Map.Entry<String, Object> entry : originalRecord.entrySet()) {
+            final String fieldName = fieldName(fieldNamePrefix, entry.getKey());
+            switch (ConnectSchema.schemaType(entry.getValue().getClass())) {


What's the possibility of a null value in an entry of such a map?

Fair point. I've added handling for nulls that just applies the null to the same field and returns.

asfbot · 2017-01-28T05:49:47Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1317/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-28T05:54:12Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1315/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-01-28T06:56:59Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1315/
Test FAILed (JDK 8 and Scala 2.12).

kkonstantine · 2017-01-28T17:39:54Z

Just a followup on the mapping's split and the unchecked exception warning.
But these are not blockers, so LGTM.

hachikuji

Minor comments in Cast. I'll get to the rest this evening.

hachikuji · 2017-01-28T22:21:03Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+            .define(SPEC_CONFIG, ConfigDef.Type.LIST, ConfigDef.NO_DEFAULT_VALUE, new ConfigDef.Validator() {
+                    @Override
+                    public void ensureValid(String name, Object valueObject) {
+                        List<String> value = (List<String>) valueObject;


nit: unchecked cast. maybe we should add @SuppressWarnings?

hachikuji · 2017-01-28T22:37:46Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+                    }
+            },
+            ConfigDef.Importance.HIGH,
+            "List of fields and the type to cast them to of the form field1:type,field2:type to cast fields of "


Do we document somewhere the list of type names that we expect or is that assumed known?

Ack, added a list here.

hachikuji · 2017-01-28T22:53:44Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+            }
+            if (parts.length == 1) {
+                if (mappings.size() > 1) {
+                    throw new ConfigException("Cast transformations that specify a type to cast the entire value to "


As far as I can tell, "int32" and ":int32" have the same effect, but the latter will skip this check since the string will split into two parts.

Hmm, interesting point. That's unintentional -- I wanted a sentinel for the single type case. I think ":int32" should probably be invalid since an empty string shouldn't be a valid field name.

In any case, in refactoring, I simplified the use of WHOLE_VALUE_CAST and it should be safe for it to now be null instead so it can't conflict with a valid field name.

hachikuji · 2017-01-28T22:59:35Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+    }
+
+    private R applySchemaless(R record) {
+        Schema.Type recordCastType = casts.get(WHOLE_VALUE_CAST);


nit: naming is a little inconsistent. Should it be something like wholeValueCastType? Similar in the case with a schema.

Updated the name

hachikuji · 2017-01-28T23:00:21Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+
+    }
+
+    private static Object castType(Object value, Schema.Type targetType) {


Maybe castTypeTo, castTo or castAs?

hachikuji · 2017-01-28T23:11:14Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+                        + " which is not supported by Connect's data API");
+            }
+            // Ensure the type we are trying to cast from is supported
+            validCastType(inferredType);


A little odd that this throws ConfigException.

ack, adjusted that method to work throw ConfigException when validating the target types and DataException on validating the input records

hachikuji · 2017-01-28T23:19:53Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+        Schema.Type recordCastType = casts.get(WHOLE_VALUE_CAST);
+
+        Schema updatedSchema = schemaUpdateCache.get(valueSchema);
+        if (updatedSchema == null) {


nit: seems we could turn this block into a function, say getOrBuildSchema?

good catch, refactored.

hachikuji · 2017-01-28T23:20:09Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+            if (valueSchema.isOptional())
+                builder.optional();
+            if (valueSchema.defaultValue() != null)
+                builder.defaultValue(valueSchema.defaultValue());


Is no conversion needed for the default value?

good catch, I got it on the fields but not on the entire record. fixed and added test.

hachikuji · 2017-01-28T23:25:29Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+        final Struct updatedValue = new Struct(updatedSchema);
+        for (Field field : value.schema().fields()) {
+            final Object origFieldValue = value.get(field);
+            updatedValue.put(updatedSchema.field(field.name()), casts.containsKey(field.name()) ? castType(origFieldValue, casts.get(field.name())) : origFieldValue);


nit: long line. Maybe we can put the second argument into a variable?

hachikuji · 2017-01-28T23:28:40Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+                        throw new ConnectException("Unexpected type in Cast transformation: " + value.getClass());
+                case BOOLEAN:
+                    if (value instanceof Number)
+                        return ((Number) value).longValue() > 0L;


Seems like we still return false on negative numbers. Did the fix get included?

hachikuji

Just a couple questions for Flatten. Looks good.

hachikuji · 2017-01-29T05:17:21Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Flatten.java

+        final Struct updatedValue = new Struct(updatedSchema);
+        buildWithSchema(value, "", updatedValue);
+        return newRecord(record, updatedSchema, updatedValue);
+


nit: unneeded newline

hachikuji · 2017-01-29T05:30:39Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Flatten.java

+                newRecord.put(fieldName(fieldNamePrefix, entry.getKey()), null);
+                return;
+            }
+            switch (ConnectSchema.schemaType(value.getClass())) {


Do we need to check the result of schemaType for null?

yeah, added the same check that's done in Cast

hachikuji · 2017-01-29T05:34:21Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Flatten.java

+                case BYTES:
+                    newRecord.put(fieldName, record.get(field));
+                    break;
+                case STRUCT:


For my own knowledge, is the MAP type only for schemaless data? You would never have a nested schemaless object inside a struct?

You can have MAP for schemas, just as you can in Avro (in fact they are a bit more flexible due to other serialization formats supporting more than just string keys). However, they can't be flattened here since we can't generate the schema without knowing what the keys will be (and in fact because keys have arbitrary structure as well and aren't necessarily just strings, it may not even be easy to generate a meaningful field name for them).

shikhar · 2017-01-29T16:42:24Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Flatten.java

+                    newSchema.field(fieldName, field.schema());
+                    break;
+                case STRUCT:
+                    buildUpdatedSchema(field.schema(), fieldName, newSchema);


I wonder if it makes sense to propagate optionality and default values when recursing. I.e. if a parent Struct was optional all the flattened fields resulting from it should be optional. And in the absence of a default value on a child field if there was a default parent Struct, use that `Struct's field value as default flattened field value.

That was a really good observation. Turns out this isn't too hard to layer onto the existing implementation, although I have to admit that I would think the semantics can be confusing to reason about. Thankfully I don't think most folks use default values on anything but individual primitive fields.

I miss @shikhar!

asfbot · 2017-02-01T02:07:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1385/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-02-01T02:07:27Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1388/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-02-01T02:12:24Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1385/
Test PASSed (JDK 7 and Scala 2.10).

xiongtx · 2017-04-14T19:01:30Z

Any updates here?

It'd be nice if KIP-66 was updated with which transformers are implemented in Kafka 0.10.2 and which aren't.

asfbot · 2017-04-14T19:04:39Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2951/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-04-14T19:04:57Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2947/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-04-14T19:05:02Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2946/
Test FAILed (JDK 7 and Scala 2.10).

ewencp · 2017-04-21T19:02:05Z

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

+
+    private static final String PURPOSE = "cast types";
+
+    private static final Set<Schema.Type> SUPPORTED_CAST_TARGETS = new HashSet<>(


Note to self: we should probably include support for casting to/from the Decimal logical type as well.

…m ancestors

ewencp · 2017-05-15T02:04:41Z

@hachikuji @kkonstantine Took another pass at this and I think I've addressed all the comments. Interesting how tricky some of these transformations can get, @shikhar's observation about optional/default values for Flatten in particular...

@xiongtx I'll update the wiki to include that info

asfbot · 2017-05-15T02:51:44Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3906/
Test FAILed (JDK 7 and Scala 2.11).

asfbot · 2017-05-15T03:23:24Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3895/
Test PASSed (JDK 8 and Scala 2.12).

ewencp · 2017-05-16T05:45:13Z

/cc @gwenshap Not sure how I omitted you thus far, pretty sure you want to see some of these transformations included :)

asfbot · 2017-05-16T06:37:13Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3978/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-16T06:47:47Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3966/
Test PASSed (JDK 8 and Scala 2.12).

kkonstantine

I see the comments have been addressed and this looks ready to me.

hachikuji

Thanks for the updates. LGTM.

kkonstantine reviewed Jan 28, 2017

View reviewed changes

hachikuji reviewed Jan 28, 2017

View reviewed changes

hachikuji reviewed Jan 29, 2017

View reviewed changes

shikhar reviewed Jan 29, 2017

View reviewed changes

ewencp force-pushed the kafka-3209-even-more-transforms branch from 9c5616d to da5f329 Compare February 1, 2017 01:19

ewencp changed the title ~~KAFKA-3209: KIP-66: Flatten and Cast single message transforms~~ KAFKA-4714: KIP-66: Flatten and Cast single message transforms Feb 1, 2017

ewencp commented Apr 21, 2017

View reviewed changes

ewencp added 5 commits May 14, 2017 11:51

KAFKA-4714: KIP-66: Flatten and Cast single message transforms

dbb8534

Update list of transformations in documentation class.

386978a

Handle null values for optional fields in Flatten transformation.

16a836d

Address review comments and checkstyle issues

ad92662

Make Flatten transformation handle optionality and default values fro…

7b23498

…m ancestors

ewencp force-pushed the kafka-3209-even-more-transforms branch from da5f329 to 7b23498 Compare May 15, 2017 02:02

ewencp changed the title ~~KAFKA-4714: KIP-66: Flatten and Cast single message transforms~~ KAFKA-4714: Flatten and Cast single message transforms (KIP-66) May 16, 2017

ewencp mentioned this pull request May 16, 2017

KAFKA-4714: TimestampConverter transformation (KIP-66) #3065

Closed

kkonstantine approved these changes May 16, 2017

View reviewed changes

hachikuji approved these changes May 17, 2017

View reviewed changes

asfgit closed this in 1cea4d8 May 17, 2017

kkonstantine added the connect label Oct 16, 2020


		}

		private static Object castType(Object value, Schema.Type targetType) {


		private static final String PURPOSE = "cast types";

		private static final Set<Schema.Type> SUPPORTED_CAST_TARGETS = new HashSet<>(

KAFKA-4714: Flatten and Cast single message transforms (KIP-66) #2458

KAFKA-4714: Flatten and Cast single message transforms (KIP-66) #2458

Conversation

ewencp commented Jan 27, 2017

ewencp commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

kkonstantine 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

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

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

asfbot commented Jan 28, 2017

asfbot commented Jan 28, 2017

asfbot commented Jan 28, 2017

kkonstantine commented Jan 28, 2017

hachikuji 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

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

hachikuji 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

Choose a reason for hiding this comment

asfbot commented Feb 1, 2017

asfbot commented Feb 1, 2017