Add Kafka headers as column #4462

0xE282B0 · 2020-07-15T19:53:48Z

This PR adds the internal _headers field to the kafka connector.
As type a map with key String and value array of byte[] is used.

For better review the changes are splited into two commits:

Extract the type definitions from the enum that describes the internal fields to a map that is initialized in the constructor where the TypeManager can be used.
Add the Header column and adapt the KafkaRecordCursor

I would appreciate if someone could review the code.

aalbu

Thank you for your PR, great addition to the connector.

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaRecordSet.java

presto-kafka/src/test/java/io/prestosql/plugin/kafka/TestKafkaIntegrationSmokeTest.java

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaRecordSet.java

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaInternalFieldDescription.java

presto-kafka/src/test/java/io/prestosql/plugin/kafka/TestKafkaIntegrationSmokeTest.java

findepi

thanks @0xE282B0 for your PR

thanks @aalbu for the review; i am just skimming at this point

presto-docs/src/main/sphinx/connector/kafka.rst

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaRecordSet.java

presto-kafka/src/test/java/io/prestosql/plugin/kafka/TestKafkaIntegrationSmokeTest.java

charlesjmorgan

Minor suggestions, thanks for your work on this :)

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaInternalFieldDescription.java

presto-kafka/src/test/java/io/prestosql/plugin/kafka/TestKafkaIntegrationSmokeTest.java

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaInternalFieldDescription.java

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaRecordSet.java

findepi · 2020-07-29T08:10:44Z

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaRecordSet.java

@@ -269,4 +278,53 @@ public void close()
            kafkaConsumer.close();
        }
    }
+
+    public static FieldValueProvider headerMapValueProvider(MapType varcharMapType, Headers headers)


I'd assume empty headers case is not a rare case, so we could want to cache a result for this (empty block).

@dain wdyt?

@0xE282B0 can you address this one unless the assumption about the fact, that empty headers is a common thing, is false?

Is there an easier way to get an empty MapBlock than creating a MapType and using a MapBlockBuilder?

MapType mapType = new MapType(VarcharType.VARCHAR, new ArrayType(VarbinaryType.VARBINARY), MethodHandles.empty(methodType(Boolean.class, Block.class, int.class, long.class)), MethodHandles.empty(methodType(Boolean.class, Block.class, int.class, Block.class, int.class)), MethodHandles.empty(methodType(long.class, Object.class)), MethodHandles.empty(methodType(long.class, Object.class))); BlockBuilder mapBlockBuilder = new MapBlockBuilder(mapType, null, 0); mapBlockBuilder.beginBlockEntry(); mapBlockBuilder.closeEntry(); Block emptyMapBlock = mapType.getObject(mapBlockBuilder, 0);

414ddcb#diff-1938e94844afe6ab2ee40936951490d9R300-R308

I am not aware. Looks good to me.

Nit: can you rename createEmptyMapBlockProvider to createEmptyHeadersFieldProvider
and EMPTY_MAP_BLOCK_PROVIDER to EMPTY_HEADERS_FIELD_PROVIDER

Also please add a test when we are reading _headers column but headers map is empty.

Naming sounds better - more specific.
The empty header test doesn't look as straightforward as I wanted it to be, but it works. Since the assertQuery does not work on maps and VARBINARY does not work with JSON, I converted the map type and then cast it to JSON to compare it with empty objects.

SELECT cast(transform_values(_headers,(k,v)->transform(v,x->from_utf8(x))) AS JSON)

Recomendations welcome 😉

I think the test could be a bit more readable if headers topic also defined named id field. It would make assertions simpler.
Then if messages with empty headers has id equal to 1 and 2.
The assertion would look like:

assertQuery("SELECT id FROM default." + headersTopic + " WHERE cardinality(_headers) = 0", "VALUES (1), (2)");

The other test query could be more readable too if you used id in WHERE clause.

WDYT?

Unfortunately we don't have an ID, but we can enumerate the messages based on the value of the message.

assertQuery("SELECT _message FROM default." + headersTopic + " WHERE cardinality(_headers) = 0", "VALUES ('1'),('2')");

I was thinking of extending the schema with id but using _message is totally fine too.

findepi · 2020-07-29T08:11:28Z

presto-kafka/src/test/java/io/prestosql/plugin/kafka/TestKafkaIntegrationSmokeTest.java

+            record = new ProducerRecord<>(topicName, null, "{}".getBytes(UTF_8));
+            record.headers()
+                    .add("foo", "bar".getBytes(UTF_8))
+                    .add("foo", null)


thanks for taking care of null values
can header key be null as well?

No, at least the Kafka Java client throws an exception if you try to create a Header with null as key.
See: https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/header/internals/RecordHeader.java#L32

losipiuk

@0xE282B0 For me the responsibility split between static enum InternalField and runtime InternalFieldDescription is very counterintuitive.

As we are forced to introduce InternalFieldDescription which is built at runtime it does not make sense to me to keep rich InternalField with subset of column information.
Instead I would suggest to move all the information to InternalFieldDescription and keep InternalField enum just as a name marker.

Given some naming changes it may look like this:

public class KafkaInternalFieldManager {
   enum InternalFieldKey {
        PARTITION_ID_FIELD,
        PARTITION_OFFSET_FIELD,
        MESSAGE_CORRUPT_FIELD,
        ...
   }

   public static class InternalField {
     private final InternalFieldKey key;
     private final String columnName;
     private final String comment;
     private final Type type;
     
     // constructor, getter, setters
  }  

  private final Map<InternalFieldKey, InternalField> internalFields;

}

We could even drop InternalFieldKey just define constants to be used for matching based on columnName.

WDYT?

0xE282B0 · 2020-08-18T20:17:04Z

Thanks for the review @losipiuk,
I agree that it feels odd to have the static enum and a representation as class with type at runtime. Therefore I'd like to get rid of the enum.
@findepi WDYT?

losipiuk

Thanks looks good. Minor comments.

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaInternalFieldManager.java

losipiuk · 2020-08-20T08:48:42Z

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaRecordSet.java

@@ -269,4 +278,53 @@ public void close()
            kafkaConsumer.close();
        }
    }
+
+    public static FieldValueProvider headerMapValueProvider(MapType varcharMapType, Headers headers)


@0xE282B0 can you address this one unless the assumption about the fact, that empty headers is a common thing, is false?

losipiuk

Thanks.
One more request for a test and we are good to go :)

losipiuk · 2020-08-31T09:48:21Z

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaRecordSet.java

@@ -269,4 +278,53 @@ public void close()
            kafkaConsumer.close();
        }
    }
+
+    public static FieldValueProvider headerMapValueProvider(MapType varcharMapType, Headers headers)


I am not aware. Looks good to me.

Nit: can you rename createEmptyMapBlockProvider to createEmptyHeadersFieldProvider
and EMPTY_MAP_BLOCK_PROVIDER to EMPTY_HEADERS_FIELD_PROVIDER

losipiuk · 2020-08-31T09:50:24Z

presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaRecordSet.java

@@ -269,4 +278,53 @@ public void close()
            kafkaConsumer.close();
        }
    }
+
+    public static FieldValueProvider headerMapValueProvider(MapType varcharMapType, Headers headers)


Also please add a test when we are reading _headers column but headers map is empty.

Signed-off-by: Sven Pfennig <[email protected]>

The KafkaInternalFieldManager creates the internalFields map in the consructor where the TypeManager can be used. Signed-off-by: Sven Pfennig <[email protected]>

Column definition has been added to KafkaInternalFieldDescription with map(VARCHAR,array(VARBINARY)) type from TypeManager. ValueProvider has been added to KafkaRecordSet Signed-off-by: Sven Pfennig <[email protected]>

losipiuk · 2020-09-10T18:41:24Z

Thanks! I did not notice previously that you already made a change to the test :)

cla-bot bot added the cla-signed label Jul 15, 2020

0xE282B0 requested review from findepi, aalbu and losipiuk July 16, 2020 19:33

findepi requested a review from charlesjmorgan July 16, 2020 20:43

aalbu reviewed Jul 16, 2020

View reviewed changes

aalbu reviewed Jul 17, 2020

View reviewed changes

presto-kafka/src/test/java/io/prestosql/plugin/kafka/TestKafkaIntegrationSmokeTest.java Show resolved Hide resolved

findepi reviewed Jul 17, 2020

View reviewed changes

charlesjmorgan reviewed Jul 17, 2020

View reviewed changes

martint requested a review from findepi July 29, 2020 06:48

findepi reviewed Jul 29, 2020

View reviewed changes

losipiuk reviewed Aug 10, 2020

View reviewed changes

0xE282B0 requested a review from losipiuk August 20, 2020 07:07

losipiuk reviewed Aug 20, 2020

View reviewed changes

losipiuk reviewed Aug 31, 2020

View reviewed changes

findepi requested a review from losipiuk September 7, 2020 10:52

Sven Pfennig added 3 commits September 9, 2020 08:56

Make Type column wider in Kafka columns documentation

3ddb4ae

Signed-off-by: Sven Pfennig <[email protected]>

Refactor KafkaInternalFieldDescription to KafkaInternalFieldManager

0daf4b1

The KafkaInternalFieldManager creates the internalFields map in the consructor where the TypeManager can be used. Signed-off-by: Sven Pfennig <[email protected]>

Add header column to Kafka Connector

228e08b

Column definition has been added to KafkaInternalFieldDescription with map(VARCHAR,array(VARBINARY)) type from TypeManager. ValueProvider has been added to KafkaRecordSet Signed-off-by: Sven Pfennig <[email protected]>

losipiuk approved these changes Sep 10, 2020

View reviewed changes

losipiuk merged commit d446959 into trinodb:master Sep 10, 2020

losipiuk mentioned this pull request Sep 10, 2020

Release notes for 342 #5111

Closed

9 tasks

martint added this to the 342 milestone Sep 24, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add Kafka headers as column #4462

Add Kafka headers as column #4462

0xE282B0 commented Jul 15, 2020 •

edited

Loading

aalbu left a comment

findepi left a comment

charlesjmorgan left a comment

findepi Jul 29, 2020

losipiuk Aug 20, 2020

0xE282B0 Aug 30, 2020

0xE282B0 Aug 30, 2020

losipiuk Aug 31, 2020

losipiuk Aug 31, 2020

0xE282B0 Sep 3, 2020

losipiuk Sep 3, 2020

0xE282B0 Sep 9, 2020

losipiuk Sep 9, 2020

findepi Jul 29, 2020

0xE282B0 Jul 29, 2020

losipiuk left a comment •

edited

Loading

0xE282B0 commented Aug 18, 2020

losipiuk left a comment

losipiuk Aug 20, 2020

losipiuk left a comment

losipiuk Aug 31, 2020

losipiuk Aug 31, 2020

losipiuk commented Sep 10, 2020

Add Kafka headers as column #4462

Add Kafka headers as column #4462

Conversation

0xE282B0 commented Jul 15, 2020 • edited Loading

aalbu left a comment

Choose a reason for hiding this comment

findepi left a comment

Choose a reason for hiding this comment

charlesjmorgan 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

losipiuk left a comment • edited Loading

Choose a reason for hiding this comment

0xE282B0 commented Aug 18, 2020

losipiuk left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

losipiuk left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

losipiuk commented Sep 10, 2020

0xE282B0 commented Jul 15, 2020 •

edited

Loading

losipiuk left a comment •

edited

Loading