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, I'd like your thoughts on the idea of incorporating the ability to skip
records with null keys. Additionally, I'm thinking of making this functionality
configurable. What are your thoughts on this?
```
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
```
default can be skipNullKeyRecords: false
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
