#21431 Pubsub dynamic topic destinations #26063

reuvenlax · 2023-04-01T00:27:18Z

No description provided.

github-actions · 2023-04-01T01:31:34Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @robertwb for label java.
R: @chamikaramj for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

reuvenlax · 2023-04-10T04:57:59Z

Run Spotless PreCommit

reuvenlax · 2023-04-11T00:18:02Z

R: @Naireen

github-actions · 2023-04-11T00:19:07Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

reuvenlax · 2023-04-15T22:06:12Z

R: @scwhittle

scwhittle · 2023-04-17T08:39:31Z

...e-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/PropertyNames.java

  public static final String PUBSUB_SUBSCRIPTION = "pubsub_subscription";
  public static final String PUBSUB_SUBSCRIPTION_OVERRIDE = "pubsub_subscription_runtime_override";
  public static final String PUBSUB_TIMESTAMP_ATTRIBUTE = "pubsub_timestamp_label";
  public static final String PUBSUB_TOPIC = "pubsub_topic";
  public static final String PUBSUB_TOPIC_OVERRIDE = "pubsub_topic_runtime_override";
+
+  public static final String PUBSUB_DYNAMIC_DESTINATIONS = "pubsub_with_dynamic_destinations";


nit: pubsub_dynamic_destinations to match enum? ie remove "with"

This has to match the string on the Dataflow backend, which is pubsub_with_dynamic_destinations

scwhittle · 2023-04-17T08:42:49Z

...e-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java

+        new PipelineVisitor.Defaults() {
+
+          @Override
+          public CompositeBehavior enterCompositeTransform(Node node) {


it should only be a composite or a primitive right?
can you simplify this test and remove the unnecessary override here?

scwhittle · 2023-04-17T08:44:41Z

...e-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java

+        new PipelineVisitor.Defaults() {
+
+          @Override
+          public CompositeBehavior enterCompositeTransform(Node node) {


scwhittle · 2023-04-17T08:45:39Z

...low-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/PubsubDynamicSink.java

+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+@SuppressWarnings({


since this is new, can you avoid supressing these?

removed the nullness. Raw types suppression has to stay, as it's due to the base class

scwhittle · 2023-04-17T08:46:55Z

...low-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/PubsubDynamicSink.java

+    }
+  }
+
+  public static class Factory implements SinkFactory {


can this be private or package visible? or perhaps even inlined in factories() above?

scwhittle · 2023-04-17T09:12:07Z

.../java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java

@@ -1350,23 +1385,15 @@ public void startBundle(StartBundleContext c) throws IOException {
      @ProcessElement
      public void processElement(ProcessContext c) throws IOException, SizeLimitExceededException {
        PubsubMessage message = getFormatFn().apply(c.element());
-        int messageSize = validateAndGetPubsubMessageSize(message);
-        if (messageSize > maxPublishBatchByteSize) {


keep this as sanity check?

scwhittle · 2023-04-17T09:16:16Z

...ogle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java

+      } else {
+        Map<TopicPath, List<OutgoingMessage>> messagesPerTopic = Maps.newHashMap();
+        for (OutgoingMessage message : messages) {
+          TopicPath topicPath = PubsubClient.topicPathFromPath(message.topic());


could do this translation during publish loop so it's per topic instead of per message

scwhittle · 2023-04-17T09:21:21Z

...ogle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java

@@ -495,6 +604,10 @@ static class PubsubSink extends PTransform<PCollection<byte[]>, PDone> {
      this.outer = outer;
    }

+    boolean isDynamic() {


remove? seems PubsubUnboundedSink would be used if this was true?

scwhittle · 2023-04-17T09:23:59Z

website/www/site/layouts/shortcodes/flink_java_pipeline_options.html

@@ -133,6 +132,11 @@
  <td>Sets the behavior of reusing objects.</td>
  <td>Default: <code>false</code></td>
 </tr>
+<tr>
+  <td><code>operatorChaining</code></td>


revert this file?

scwhittle · 2023-04-17T09:24:22Z

website/www/site/layouts/shortcodes/flink_python_pipeline_options.html

@@ -133,6 +132,11 @@
  <td>Sets the behavior of reusing objects.</td>
  <td>Default: <code>false</code></td>
 </tr>
+<tr>
+  <td><code>operator_chaining</code></td>


this file still seems there

scwhittle · 2023-04-19T09:30:31Z

...low-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/PubsubDynamicSink.java

+  private final String timestampLabel;
+  private final String idLabel;
+  private final StreamingModeExecutionContext context;
+  // Function used to convert PCollection elements to PubsubMessage objects.


remove comment?

scwhittle · 2023-04-19T10:03:33Z

...low-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/PubsubDynamicSink.java

+    public ByteString getDataFromMessage(PubsubMessage formatted, ByteStringOutputStream stream)
+        throws IOException {
+      Pubsub.PubsubMessage.Builder pubsubMessageBuilder =
+          Pubsub.PubsubMessage.newBuilder().setData(ByteString.copyFrom(formatted.getPayload()));


I wonder if we should separately look into switching to ByteString within beam's PubsubMessage.

scwhittle · 2023-04-19T10:05:52Z

...low-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/PubsubDynamicSink.java

+    @Override
+    public long add(WindowedValue<PubsubMessage> data) throws IOException {
+      String dataTopic =
+          org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull(


import static?

scwhittle · 2023-04-19T10:09:28Z

...java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/PubsubDynamicSinkTest.java

+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.when;
+
+import avro.shaded.com.google.common.collect.Lists;


don't use shaded version?

scwhittle · 2023-04-19T10:18:25Z

...ogle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java

        if (!pubsubMessages.isEmpty()
-            && bytes + message.message().getData().size() > publishBatchBytes) {
+            && bytes + message.getMessage().getData().size() > publishBatchBytes) {


this limit is currently across all topics but could be per-topic

another option would be to change so we group by topic+shard instead of just shard
that would keep this simple here, and avoid sequential blocking publishes

Added grouping by topic+shard

scwhittle · 2023-04-19T10:19:21Z

...ogle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java

+      // dynamic destinations.
+      return input
+          .apply(
+              "WithDynamicKeys",


nit: WithDynamicTopics? DynamicDestinations?
keys seems implementation detail

scwhittle · 2023-04-19T10:26:09Z

website/www/site/layouts/shortcodes/flink_python_pipeline_options.html

@@ -133,6 +132,11 @@
  <td>Sets the behavior of reusing objects.</td>
  <td>Default: <code>false</code></td>
 </tr>
+<tr>
+  <td><code>operator_chaining</code></td>


this file still seems there

scwhittle · 2023-04-19T10:35:44Z

.../java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java

      switch (input.isBounded()) {
        case BOUNDED:
-          input.apply(
+          pubsubMessages.apply(


enforce no dynamic destinations in batch? Or I think the BoundedWriter needs to group output by topic as publish is just all messages to single topic currently.

good point - fixed BoundedWriter

scwhittle · 2023-04-20T08:25:39Z

...a/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java

+  }
+
+  public void testDynamicTopics(boolean isBounded) throws IOException {
+    List<OutgoingMessage> outgoing =


nit: name expectedOutgoing, it wasn't obvious that the test factory for publish took these as expectations when created. I think naming would help.

scwhittle · 2023-04-20T08:31:21Z

.../java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java

@@ -1355,9 +1377,10 @@ public void populateDisplayData(DisplayData.Builder builder) {
     * <p>Public so can be suppressed by runners.
     */
    public class PubsubBoundedWriter extends DoFn<PubsubMessage, Void> {
-      private transient List<OutgoingMessage> output;
+      private transient Map<PubsubTopic, List<OutgoingMessage>> output;


could have a single map by topic, reduces lookup and prevents possibility of entry in one map and not the other.

scwhittle · 2023-04-20T08:32:50Z

.../java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java

        pubsubClient.close();
        pubsubClient = null;
      }

      private void publish() throws IOException {
-        PubsubTopic topic = getTopicProvider().get();
+        for (Map.Entry<PubsubTopic, List<OutgoingMessage>> entry : output.entrySet()) {


use visitor instead of iterator?
forEach(e -> publish(e.getKey(), e.getValue());

Unfortunately this does not work for function references that throw exceptions. I would have to wrap a try/catch inside the lambda (i.e. catch the IOException and rethrow as a RuntimeException), which gets ugly.

scwhittle · 2023-04-20T08:35:02Z

.../java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java

+          publish(pubsubTopic, currentTopicOutput);
+          currentTopicOutput.clear();
+          currentTopicOutputBytes = 0;
+          currentOutputBytes.put(pubsubTopic, 0L);


think the single map would make this simpler to since you would be updating the value in the entry directly

scwhittle · 2023-04-20T08:35:48Z

...ud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithTopicCoder.java

+  // A message's attributes can be null.
+  private static final Coder<Map<String, String>> ATTRIBUTES_CODER =
+      NullableCoder.of(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()));
+  // A message's messageId cannot be null


"cannot be null" is inconsistent with using NullableCoder

scwhittle · 2023-04-20T08:36:43Z

...ud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithTopicCoder.java

+  @Override
+  public PubsubMessage decode(InputStream inStream) throws IOException {
+    byte[] payload = PAYLOAD_CODER.decode(inStream);
+    Map<String, String> attributes = ATTRIBUTES_CODER.decode(inStream);


add Nullable annotation?

scwhittle · 2023-04-20T08:37:07Z

...ud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithTopicCoder.java

+
+/** A coder for PubsubMessage including the topic from the PubSub server. */
+@SuppressWarnings({
+  "nullness" // TODO(https://github.com/apache/beam/issues/20497)


can this be removed?

reuvenlax · 2023-04-20T17:56:31Z

Run Java PreCommit

scwhittle

LGTM with one comment

scwhittle · 2023-04-20T18:32:03Z

.../java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java

      }

      @FinishBundle
      public void finishBundle() throws IOException {
        publish();
        output = null;
-        currentOutputBytes = null;
        pubsubClient.close();
        pubsubClient = null;
      }

      private void publish() throws IOException {


inline this in finishBundle? otherwise the topic specific publish should be called

…ent in apache/beam#26063. PiperOrigin-RevId: 528517450

kennknowles · 2024-03-11T23:53:26Z

In authoring #30604 some test from this PR failed. Specifically something that appears to expect client-side overrides (aka the v1 path) to have happened when using v2. Does that make sense to you all?

kennknowles · 2024-03-11T23:55:09Z

...e-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java

+            }
+          }
+        });
+    assertTrue(sawPubsubOverride.get());


We should never be using client-side overrides for v2, where we want it all to happen in the service. Is there necessary functionality in the override? It happens after the v2 pipeline is created anyhow I think...

reuvenlax · 2024-03-12T00:03:49Z

If you are talking about Dataflow runner v2, those overrides were implemented service side not client side.

…

On Mon, Mar 11, 2024 at 4:53 PM Kenn Knowles ***@***.***> wrote: In authoring #30604 <#30604> some test from this PR failed. Specifically something that appears to expect client-side overrides (aka the v1 path) to have happened when using v2. Does that make sense to you all? — Reply to this email directly, view it on GitHub <#26063 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFAYJVMKPVSUBYVUWWSVE53YXZGZZAVCNFSM6AAAAAAWPIJDGCVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMYTSOBZGY2TSMRQGU> . You are receiving this because you modified the open/close state.Message ID: ***@***.***>

github-actions bot added dataflow gcp io java runners labels Apr 1, 2023

github-actions bot added Next Action: Reviewers core and removed core labels Apr 1, 2023

github-actions bot added core and removed core labels Apr 8, 2023

reuvenlax force-pushed the pubsub_dynamic_destinations branch from ba4dd0b to e11a5ac Compare April 9, 2023 18:07

github-actions bot added core and removed core labels Apr 9, 2023

reuvenlax changed the title ~~Pubsub dynamic destinations~~ #21431 Pubsub dynamic topic destinations Apr 11, 2023

reuvenlax force-pushed the pubsub_dynamic_destinations branch from 96ed323 to 39b56fc Compare April 11, 2023 00:17

github-actions bot added the core label Apr 11, 2023

github-actions bot removed the core label Apr 11, 2023

github-actions bot added core website labels Apr 12, 2023

scwhittle requested changes Apr 17, 2023

View reviewed changes

reuvenlax force-pushed the pubsub_dynamic_destinations branch from 6660fdc to b394cf4 Compare April 18, 2023 19:33

scwhittle requested changes Apr 19, 2023

View reviewed changes

github-actions bot removed the website label Apr 19, 2023

scwhittle requested changes Apr 20, 2023

View reviewed changes

reuvenlax force-pushed the pubsub_dynamic_destinations branch from 7809f08 to 1cb9649 Compare April 22, 2023 01:48

Pubsub dynamic topic destinations

ca67490

reuvenlax force-pushed the pubsub_dynamic_destinations branch from 1cb9649 to ca67490 Compare April 22, 2023 02:16

reuvenlax added 2 commits April 21, 2023 19:44

foo

55e5932

update vendored dependency

f592806

reuvenlax merged commit f549fd3 into apache:master Apr 22, 2023

copybara-service bot pushed a commit to GoogleCloudPlatform/DataflowTemplates that referenced this pull request May 1, 2023

Quick fix for a syndeo build issue after a change in Beam's PubsubCli…

4928373

…ent in apache/beam#26063. PiperOrigin-RevId: 528517450

chamikaramj mentioned this pull request Jun 22, 2023

[Bug]: SizeLimitExceededException thrown while sending a small message to GCP PubSub #27178

Closed

15 tasks

kennknowles reviewed Mar 11, 2024

View reviewed changes

#21431 Pubsub dynamic topic destinations #26063

#21431 Pubsub dynamic topic destinations #26063

Conversation

reuvenlax commented Apr 1, 2023

github-actions bot commented Apr 1, 2023

reuvenlax commented Apr 10, 2023

reuvenlax commented Apr 11, 2023

github-actions bot commented Apr 11, 2023

reuvenlax commented Apr 15, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

reuvenlax commented Apr 20, 2023

scwhittle left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kennknowles commented Mar 11, 2024

Choose a reason for hiding this comment

reuvenlax commented Mar 12, 2024 via email