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

[GitBox <gi...@apache.org>]

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



##########
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:
       For offset topic, the key must not be `null`, which is guaranteed by Kafka itself. Kafka's coordinator writes messages that must have keys to the offset topic. In addition, Kafka broker prevents any write to the offset topic by a Kafka client.
   
   Therefore, I think there is no need to perform null check here. If you insisted to check null keys, the logs here should be error because it's an unexpected case.

##########
File path: pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarOffsetBackingStore.java
##########
@@ -49,7 +50,7 @@
 @Slf4j
 public class PulsarOffsetBackingStore implements OffsetBackingStore {
 
-    private Map<ByteBuffer, ByteBuffer> data;
+    private ConcurrentHashMap<ByteBuffer, ByteBuffer> data;

Review comment:
       I think it's better to make it `final`. And there is no need to change the type to `ConcurrentHashMap` because we only use the common methods in `Map`.




-- 
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-unsubscribe@pulsar.apache.org

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