lukecwik commented on a change in pull request #12806:
URL: https://github.com/apache/beam/pull/12806#discussion_r488216535
##########
File path:
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
##########
@@ -1010,20 +1011,27 @@ public PDone expand(PCollection<T> input) {
getMaxBatchBytesSize(),
MAX_PUBLISH_BATCH_BYTE_SIZE_DEFAULT))));
return PDone.in(input.getPipeline());
case UNBOUNDED:
- return input
- .apply(MapElements.into(new TypeDescriptor<PubsubMessage>()
{}).via(getFormatFn()))
- .apply(
- new PubsubUnboundedSink(
- getPubsubClientFactory(),
- NestedValueProvider.of(getTopicProvider(), new
TopicPathTranslator()),
- getTimestampAttribute(),
- getIdAttribute(),
- 100 /* numShards */,
- MoreObjects.firstNonNull(
- getMaxBatchSize(),
PubsubUnboundedSink.DEFAULT_PUBLISH_BATCH_SIZE),
- MoreObjects.firstNonNull(
- getMaxBatchBytesSize(),
- PubsubUnboundedSink.DEFAULT_PUBLISH_BATCH_BYTES)));
+ PCollection<PubsubMessage> output =
+ input.apply(
+ MapElements.into(new TypeDescriptor<PubsubMessage>()
{}).via(getFormatFn()));
+ if
(ExperimentalOptions.hasExperiment(input.getPipeline().getOptions(),
"beam_fn_api")
Review comment:
I think this should be "beam_fn_api" || "use_runner_v2" ||
"use_unified_worker"
Also, can we use the full pubsub with message attributes coder, this will
allow for future changes to the PubsubUnboundedSink without needing to break
update compatibility
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
