ahmedabu98 commented on code in PR #31608:
URL: https://github.com/apache/beam/pull/31608#discussion_r1759246839
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java:
##########
@@ -2033,6 +2034,16 @@ private static void translate(
PubsubUnboundedSink overriddenTransform,
StepTranslationContext stepContext,
PCollection input) {
+ if (overriddenTransform.getPublishBatchWithOrderingKey()) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "%s does not currently support publishing with ordering keys. "
Review Comment:
Done
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PreparePubsubWriteDoFn.java:
##########
@@ -66,6 +68,20 @@ static int validatePubsubMessageSize(PubsubMessage message,
int maxPublishBatchS
}
int totalSize = payloadSize;
+ @Nullable String orderingKey = message.getOrderingKey();
+ if (orderingKey != null) {
+ int orderingKeySize =
orderingKey.getBytes(StandardCharsets.UTF_8).length;
+ if (orderingKeySize > PUBSUB_MESSAGE_ATTRIBUTE_MAX_VALUE_BYTES) {
Review Comment:
Done
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PreparePubsubWriteDoFn.java:
##########
@@ -175,6 +193,16 @@ public void process(
.add("pubsub", "topic",
PubsubClient.topicPathFromPath(topic).getDataCatalogSegments());
reportedLineage = topic;
}
+ // TODO: Remove this check once Dataflow's native sink supports ordering
keys.
Review Comment:
@sjvanrossum wdyt? I raised the same concern in the other
[thread](https://github.com/ahmedabu98/beam/pull/427#discussion_r1650304238)
and am still leaning in that direction
--
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: github-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
