PubSub Sink: Empty (non-null) message values bypass null check and break processing

[FireBurn]

Environment details

1. API: Pub/Sub Kafka Connector (Sink)
2. OS type and version: RHEL 8.10
3. Java version: Amazon Corretto 21.0.11.10.1 (build 21.0.11+10-LTS)
4. Connector version(s): 1.3.3

Steps to reproduce

1. Produce a Kafka message where:
   • value = "" (empty string, not null)
   • no headers/attributes are present
2. Allow the Pub/Sub Sink Connector to consume the message
3. Observe connector behaviour

Expected behaviour

Messages with no meaningful content (null OR empty string) should be treated as invalid and skipped.

The connector should continue processing subsequent messages without interruption.

Actual behaviour

When a message contains an empty string ("") rather than null, it bypasses the existing check:

if (attributes.size() == 0 && value == null)

This results in:

• The message being treated as valid
• Downstream processing issues
• Connector processing stopping or stalling

Code example

Current implementation:

if (attributes.size() == 0 && value == null) {
  log.warn("Message received with no value and no attributes. Not publishing message");
  SettableApiFuture<String> nullMessageFuture = SettableApiFuture.<String>create();
  nullMessageFuture.set("No message");
  addPendingMessageFuture(record.topic(), record.kafkaPartition(), nullMessageFuture);
  continue;
}

Proposed fix

Update the condition to treat empty strings as equivalent to null:

--- src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkTask.java
+++ src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkTask.java
@@ -166,7 +166,7 @@
           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");

Rationale

An empty string ("") is effectively equivalent to a missing payload in this context.

The current implementation only checks for null, allowing empty-but-invalid messages through, which can lead to inconsistent behaviour and connector stalls.

Handling both null and empty values consistently would improve robustness.

Stack trace

No stack trace available.

Additional information

This has been observed in environments where upstream producers occasionally emit empty string payloads instead of null values.

[FireBurn]

#431