googleapis / googleapis/java-pubsub-group-kafka-connector
PubSub Sink: Empty (non-null) message values bypass null check and break processing
- Dominant language
- Java
- Stars
- 50
- Forks
- 37
- Avg merge
- 1d 19h
- Merged PRs (30d)
- 3
Description
#### 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 nullMessageFuture = SettableApiFuture.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 nullMessageFuture = SettableApiFuture.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.
Contributor guide
Research direction
Start in src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkTask.java at the message validation condition around line 166. Review how records with no attributes and null values are skipped, then verify that empty-string values follow the same path. Done means both null and empty values are skipped and subsequent messages continue processing.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- backend
- Issue type
- Bug
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 74/100