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

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

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



##########
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:
       The messages managed by the backing store service itself do have the guarantees, but we do not have the guarantees from the user themselves, if the user send test messages to the offset topic (i.e. use `pulsar-client produce`) and it could cause the unexpected issues. With this scenario, the unexpected messages should be ignored and not affect the backing store, and this is why it been log as debug. 




-- 
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