Add kafka source with additional variable with metadata. #1512

gskrobisz · 2021-04-08T16:58:12Z

No description provided.

...in/scala/pl/touk/nussknacker/engine/kafka/consumerrecord/ConsumerRecordToJsonFormatter.scala

mproch · 2021-04-12T15:00:16Z

engine/flink/kafka-util/src/main/scala/pl/touk/nussknacker/engine/kafka/source/InputMeta.scala

+    TypedObjectTypingResult(fields(keyTypingResult), objType(keyTypingResult))
+  }
+
+  private def fields(keyTypingResult: typing.TypingResult): Map[String, TypingResult] = {


maybe we could just introduce case class here?

We already have InputMeta case class. The problem here is that we mix type of key with primitive fields. IMO this way is ok, to resolve this problem. This companion object is in the same file as class so probability of desynchornization is minimal.

Fixed (and introduced case class SerializableConsumerRecord)

...l/src/main/scala/pl/touk/nussknacker/engine/kafka/source/KafkaGenericNodeSourceFactory.scala

...k/kafka-util/src/main/scala/pl/touk/nussknacker/engine/kafka/source/KafkaSourceFactory.scala

...ouk/nussknacker/engine/kafka/consumerrecord/ConsumerRecordDeserializationSchemaFactory.scala

...k/kafka-util/src/main/scala/pl/touk/nussknacker/engine/kafka/source/KafkaSourceFactory.scala

arkadius · 2021-05-05T14:07:17Z

...in/scala/pl/touk/nussknacker/engine/kafka/consumerrecord/ConsumerRecordToJsonFormatter.scala

+
+  def from[K, V](topic: String, record: SerializableConsumerRecord[K, V]): ConsumerRecord[K, V] = {
+    createConsumerRecord(
+      record.topic.getOrElse(topic),


It is counter-intuitive that topic specified in record has higher priority than record.topic. WHy this method is needed? Why not to just serializableConsumerRecord.copy(topic = Some(otherTopic)).toConsumerRecord)?

Also I see now that this method has cheated me - take a look for full invocation: looking on SerializableConsumerRecord from (....) you will expect that the result will be SerializableConsumerRecord, not kafka ConsumerRecord

Fixed according to our discussion.

...ouk/nussknacker/engine/kafka/consumerrecord/ConsumerRecordDeserializationSchemaFactory.scala

arkadius · 2021-05-05T14:25:21Z

...in/scala/pl/touk/nussknacker/engine/kafka/consumerrecord/ConsumerRecordToJsonFormatter.scala

+
+object SerializableConsumerRecord {
+
+  def createConsumerRecord[K, V](topic: String, partition: Int, offset: Long, timestamp: Long, key: K, value: V, headers: Headers): ConsumerRecord[K, V] = {


Am I right that it is only used on data extracted from SerializableConsumerRecord? If it so, why not too have just a method toKafkaConsumerRecord inside SerializableConsumerRecord.

Fixed, according to our discussion.

...in/scala/pl/touk/nussknacker/engine/kafka/consumerrecord/ConsumerRecordToJsonFormatter.scala

engine/flink/kafka-util/src/main/scala/pl/touk/nussknacker/engine/kafka/generic/sources.scala

arkadius · 2021-05-12T08:26:22Z

docs/MigrationGuide.md

+  - current `RecordFormater` should be sufficient for value-only serialization, or use `ConsumerRecordToJsonFormatter` for metadata serialization
+  - provide timestampAssigner that is able to extract time from `ConsumerRecord[K, V]`
+  Also:
+  - removed `BaseKafkaSourceFactory` with multiple topics support: use `KafkaSourceFactory` instead, see test "source with two input topics"


can you fix the order? should it be see "source with two input topics" test case

arkadius

LGTM, have some comments to type info detection because I didn't looked into it so far, maybe @mproch also take a look on it

arkadius · 2021-05-17T15:41:41Z

...in/scala/pl/touk/nussknacker/engine/flink/api/typeinformation/TypeInformationDetection.scala

+/**
+  * Trait that allows for providing more details TypeInformation when TypingResult is known.
+  */
+trait TypeInformationDetectionForTypingResult extends TypeInformationDetection {


Why not to merge it with TypeInformationDetection? 2. Maybe it could have better name because now it looks like it only detect type for typing result (fot context and so on - no)

👍 If it extends TypeInformationDetection I think it's better to move this method to forType?

Fixed. Removed TypeInformationDetectionForTypingResult, method forType moved to TypeInformationDetection.

arkadius · 2021-05-17T15:46:00Z

engine/flink/kafka-util/src/main/scala/pl/touk/nussknacker/engine/kafka/source/InputMeta.scala

+/**
+  * Customisation for TypeInformationDetection that provides type information for BaseInputMeta.
+  */
+class InputMetaAwareTypeInformationCustomisation extends TypingResultAwareTypeInformationCustomisation {


Shouldn't it be used in generic/demo/dev model?

👍 Currently it does not seem to be used anywhere? At least we should register it with ServiceLoader?

Fixed. See META-INF/services in flinkKafkaUtil and process. Example test scenario is in InputMetaDeserializationSpec.

mproch · 2021-05-07T05:07:33Z

...ouk/nussknacker/engine/kafka/consumerrecord/ConsumerRecordDeserializationSchemaFactory.scala

+  * @tparam K - type of key of deserialized ConsumerRecord
+  * @tparam V - type of value of deserialized ConsumerRecord
+  */
+@silent("deprecated")


what deprecation warnings are here? Can we do sth to avoid them?

Fixed. ConsumerRecord marks field checksum as deprecated. We do not need to use this field, therefore here I replaced it with default NULL_CHECKSUM value.

mproch · 2021-05-18T08:54:25Z

...in/scala/pl/touk/nussknacker/engine/flink/api/typeinformation/TypeInformationDetection.scala

+/**
+  * Trait that allows for providing more details TypeInformation when TypingResult is known.
+  */
+trait TypeInformationDetectionForTypingResult extends TypeInformationDetection {


👍 If it extends TypeInformationDetection I think it's better to move this method to forType?

mproch · 2021-05-18T08:56:49Z

engine/flink/kafka-util/src/main/scala/pl/touk/nussknacker/engine/kafka/source/InputMeta.scala

+/**
+  * Customisation for TypeInformationDetection that provides type information for BaseInputMeta.
+  */
+class InputMetaAwareTypeInformationCustomisation extends TypingResultAwareTypeInformationCustomisation {


👍 Currently it does not seem to be used anywhere? At least we should register it with ServiceLoader?

…ypeInformationDetection

gskrobisz linked an issue Apr 8, 2021 that may be closed by this pull request

Source should be able to control start variables #548

Closed

gskrobisz force-pushed the feature/kafka_source_with_metadata branch 3 times, most recently from 997596f to 444c8b4 Compare April 12, 2021 13:32

gskrobisz changed the base branch from improvements/flinksource_api_refactoring to staging April 12, 2021 13:33

gskrobisz changed the title ~~Add kafka source that allows additional variable with metadata.~~ Add kafka source with additional variable with metadata. Apr 12, 2021

mproch reviewed Apr 12, 2021

View reviewed changes

arkadius reviewed Apr 13, 2021

View reviewed changes

gskrobisz force-pushed the feature/kafka_source_with_metadata branch 3 times, most recently from eca25b1 to 8a024a9 Compare May 5, 2021 11:49

arkadius reviewed May 5, 2021

View reviewed changes

...k/kafka-util/src/main/scala/pl/touk/nussknacker/engine/kafka/source/KafkaSourceFactory.scala Show resolved Hide resolved

arkadius reviewed May 5, 2021

View reviewed changes

...ouk/nussknacker/engine/kafka/consumerrecord/ConsumerRecordDeserializationSchemaFactory.scala Show resolved Hide resolved

arkadius reviewed May 5, 2021

View reviewed changes

...in/scala/pl/touk/nussknacker/engine/kafka/consumerrecord/ConsumerRecordToJsonFormatter.scala Outdated Show resolved Hide resolved

arkadius reviewed May 5, 2021

View reviewed changes

engine/flink/kafka-util/src/main/scala/pl/touk/nussknacker/engine/kafka/generic/sources.scala Outdated Show resolved Hide resolved

gskrobisz force-pushed the feature/kafka_source_with_metadata branch 2 times, most recently from 4e830f2 to 787efed Compare May 11, 2021 15:03

arkadius reviewed May 12, 2021

View reviewed changes

gskrobisz added 8 commits May 14, 2021 13:21

Add kafka source with additional variable with metadata.

b0e736e

CR fixes

0b4db15

CR fixes part 2, TypeInformation for InputMeta

546fef3

CR fixes part 3, improved SerializableConsumerRecord

a07741c

CR fixes part 4, fixed hints for #inputMeta properties

a58fb8e

CR fixes part 5, fixed TypingResult definition for #inputMeta.headers

7ea1d56

CR fixes part 6, extractor for BaseInputMetaVariables

3e6c864

CR fixes part 7, display correct #inputMeta type

f19eb86

gskrobisz force-pushed the feature/kafka_source_with_metadata branch from a449a8e to f19eb86 Compare May 14, 2021 11:47

CR fixes part 8, rollback to explicite field list in InputMeta

2cbe88f

arkadius approved these changes May 17, 2021

View reviewed changes

mproch reviewed May 18, 2021

View reviewed changes

CR fixes part 9, merge TypeInformationDetectionForTypingResult with T…

77906d0

…ypeInformationDetection

gskrobisz merged commit a746020 into staging May 18, 2021

dswiecki deleted the feature/kafka_source_with_metadata branch September 7, 2021 08:33

snyk-bot mentioned this pull request Sep 21, 2021

[Snyk] Security upgrade jointjs from 3.0.2 to 3.4.2 lciolecki/nussknacker#12

Open

This was referenced Oct 6, 2022

[Snyk] Fix for 20 vulnerabilities lciolecki/nussknacker#18

Open

[Snyk] Fix for 20 vulnerabilities lciolecki/nussknacker#19

Open

[Snyk] Fix for 20 vulnerabilities lciolecki/nussknacker#20

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add kafka source with additional variable with metadata. #1512

Add kafka source with additional variable with metadata. #1512

gskrobisz commented Apr 8, 2021

mproch Apr 12, 2021

arkadius Apr 13, 2021

gskrobisz May 5, 2021

arkadius May 5, 2021

arkadius May 5, 2021

gskrobisz May 13, 2021

arkadius May 5, 2021

gskrobisz May 13, 2021

arkadius May 12, 2021

gskrobisz May 13, 2021

arkadius left a comment

arkadius May 17, 2021

mproch May 18, 2021

gskrobisz May 18, 2021

arkadius May 17, 2021

mproch May 18, 2021

gskrobisz May 18, 2021

mproch May 7, 2021

gskrobisz May 18, 2021

mproch May 18, 2021

mproch May 18, 2021


		object SerializableConsumerRecord {

		def createConsumerRecord[K, V](topic: String, partition: Int, offset: Long, timestamp: Long, key: K, value: V, headers: Headers): ConsumerRecord[K, V] = {

Add kafka source with additional variable with metadata. #1512

Add kafka source with additional variable with metadata. #1512

Conversation

gskrobisz commented Apr 8, 2021

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

arkadius 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