[GitHub] [pulsar] BewareMyPower commented on a change in pull request #14491: [pulsar-io] throw exceptions when kafka offset backing store failed to start

[GitBox]

BewareMyPower commented on a change in pull request #14491:
URL: https://github.com/apache/pulsar/pull/14491#discussion_r816518706



##########
File path: 
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarOffsetBackingStore.java
##########
@@ -126,10 +127,13 @@ private void readNext(CompletableFuture<Void> endFuture) {
     }
 
     void processMessage(Message<byte[]> message) {
-        synchronized (data) {
+        if (message.getKey() != null) {
             data.put(
                 ByteBuffer.wrap(message.getKey().getBytes(UTF_8)),
                 ByteBuffer.wrap(message.getValue()));
+        } else {
+            log.debug("Got message without key from the offset storage topic, 
skip it. message value: {}",

Review comment:
       It looks like the topic is produced by Pulsar producer. See `set` method:
   
   ```java
               producer.newMessage()
                   .key(new String(keyBytes, UTF_8))
   ```
   
   So I think the key could never be null.




-- 
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: commits-unsubscr...@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org