stankiewicz commented on code in PR #39444:
URL: https://github.com/apache/beam/pull/39444#discussion_r3660078685
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubJsonClient.java:
##########
@@ -181,45 +178,51 @@ public List<IncomingMessage> pull(
new
PullRequest().setReturnImmediately(returnImmediately).setMaxMessages(batchSize);
PullResponse response =
pubsub.projects().subscriptions().pull(subscription.getPath(),
request).execute();
- if (response.getReceivedMessages() == null ||
response.getReceivedMessages().isEmpty()) {
+ List<ReceivedMessage> receivedMessages = response.getReceivedMessages();
+ if (receivedMessages == null || receivedMessages.isEmpty()) {
return ImmutableList.of();
}
- List<IncomingMessage> incomingMessages = new
ArrayList<>(response.getReceivedMessages().size());
- for (ReceivedMessage message : response.getReceivedMessages()) {
+ List<IncomingMessage> incomingMessages = new
ArrayList<>(receivedMessages.size());
+ for (ReceivedMessage message : receivedMessages) {
PubsubMessage pubsubMessage = message.getMessage();
- Map<String, String> attributes;
- if (pubsubMessage.getAttributes() != null) {
- attributes = pubsubMessage.getAttributes();
- } else {
+ if (pubsubMessage == null) {
+ continue;
+ }
+ Map<String, String> attributes = pubsubMessage.getAttributes();
+ if (attributes == null) {
attributes = new HashMap<>();
}
// Payload.
- byte[] elementBytes = pubsubMessage.getData() == null ? null :
pubsubMessage.decodeData();
- if (elementBytes == null) {
- elementBytes = new byte[0];
- }
+ String dataStr = pubsubMessage.getData();
+ byte[] elementBytes = (dataStr == null) ? new byte[0] :
pubsubMessage.decodeData();
// Timestamp.
long timestampMsSinceEpoch;
- if (Strings.isNullOrEmpty(timestampAttribute)) {
- timestampMsSinceEpoch =
parseTimestampAsMsSinceEpoch(message.getMessage().getPublishTime());
+ if (timestampAttribute == null || timestampAttribute.isEmpty()) {
+ String publishTime = pubsubMessage.getPublishTime();
+ if (publishTime == null) {
+ throw new IllegalStateException("Received message missing
publishTime");
+ }
+ timestampMsSinceEpoch = parseTimestampAsMsSinceEpoch(publishTime);
} else {
timestampMsSinceEpoch = extractTimestampAttribute(timestampAttribute,
attributes);
}
// Ack id.
String ackId = message.getAckId();
- checkState(!Strings.isNullOrEmpty(ackId));
+ if (ackId == null || ackId.isEmpty()) {
+ throw new IllegalStateException("Received message missing ackId");
+ }
// Record id, if any.
@Nullable String recordId = null;
if (idAttribute != null) {
recordId = attributes.get(idAttribute);
}
- if (Strings.isNullOrEmpty(recordId)) {
Review Comment:
why to change?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
