fix: handle empty string message values in CloudPubSubSinkTask (#431)

Treat empty string ("") payloads as equivalent to null when no attributes are present. Previously, only null values were checked, allowing empty messages to pass through and potentially stall processing.

Updated the condition to use attributes.isEmpty() and include value.isEmpty() alongside the null check to ensure invalid messages are consistently skipped.

Author: FireBurn

src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkTask.java

@@ -176,7 +176,7 @@ public void put(Collection<SinkRecord> sinkRecords) {
           attributes.put(header.key(), header.value().toString());
         }
       }
-      if (attributes.size() == 0 && value == null) {
+      if (attributes.isEmpty() && (value == null || value.isEmpty())) {
         log.warn("Message received with no value and no attributes. Not publishing message");
         SettableApiFuture<String> nullMessageFuture = SettableApiFuture.<String>create();
         nullMessageFuture.set("No message");