swamymavuri commented on code in PR #46:
URL: https://github.com/apache/pulsar-adapters/pull/46#discussion_r1146162466
##########
pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java:
##########
@@ -404,10 +409,28 @@ public ConsumerRecords<K, V> poll(long timeoutMillis) {
timestamp = msg.getEventTime();
timestampType = TimestampType.CREATE_TIME;
}
-
- ConsumerRecord<K, V> consumerRecord = new
ConsumerRecord<>(topic, partition, offset, timestamp,
- timestampType, -1, msg.hasKey() ?
msg.getKey().length() : 0, msg.getData().length, key, value);
-
+ ConsumerRecord<K, V> consumerRecord;
+ if (msg.getProperties() != null) {
+ Headers headers = new RecordHeaders();
+ msg.getProperties().forEach((k, v) -> {
+ if
(k.startsWith(MessageConstants.KAFKA_MESSAGE_HEADER_PREFIX)) {
Review Comment:
Here, we are filtering header keys other than starting with **kafka.header.**
But in pulsar, every property added to TypedMessageBuilder is treated as
header(for ex.Partitioned id ). Here, we are skipping those properties as
those variables are not encoded in Hex and converting those will lead to
**Decoder exception.**
if we want those properties also, we need to add one more condition as
below.
```
if (k.startsWith(MessageConstants.KAFKA_MESSAGE_HEADER_PREFIX)) {
headers.add(originalKey, Hex.decodeHex(v));
}else{
headers.add(originalKey, v.getBytes());
}
```
@lhotari @cbornet @dlg99
Let me know is to **OK** to skip those property headers?
--
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]
