kazimirov999 commented on code in PR #336:
URL: https://github.com/apache/flink-statefun/pull/336#discussion_r1392929689
##########
statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/binders/ingress/v1/RoutableKafkaIngressDeserializer.java:
##########
@@ -45,7 +45,7 @@ public RoutableKafkaIngressDeserializer(Map<String,
RoutingConfig> routingConfig
public Message deserialize(ConsumerRecord<byte[], byte[]> input) {
final String topic = input.topic();
final byte[] payload = input.value();
- final byte[] key = requireNonNullKey(input.key());
Review Comment:
hi, what do you think about adding the ability to skip the record with null
key and make this functionality configurable like:
```
kind: io.statefun.kafka.v1/ingress
spec:
id: localhost:9092
consumerGroupId: test-group-dev
startupPosition:
type: earliest
properties:
- isolation.level: read_committed
- security.protocol: SSL
- ssl.truststore.location: /tmp/flink/jks/truststore.jks
- ssl.truststore.password: changeit
- ssl.keystore.location: /tmp/flink/jks/kafka_keystore.jks
- ssl.keystore.password: test
- ssl.key.password: test
topics:
- topic: test-dev
skipNullKeyRecords: true
valueType: ua.Test
targets:
- ua.Test
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
