googleapis / googleapis/java-pubsub-group-kafka-connector

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

Open Beginner friendly
#430 1 comment 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.