[GitHub] [beam] zhoufek commented on a change in pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage

[GitBox <gi...@apache.org>]

zhoufek commented on a change in pull request #14971:
URL: https://github.com/apache/beam/pull/14971#discussion_r648482674



##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
##########
@@ -483,6 +488,54 @@ public String toString() {
     return Read.newBuilder(parsePayloadUsingCoder(coder)).setCoder(coder).build();
   }
 
+  /**
+   * Returns a {@link PTransform} that continuously reads binary encoded protobuf messages for the
+   * type specified by {@code fullMessageName}.
+   *
+   * <p>This is primarily here for cases where the message type cannot be known at compile time. If
+   * it can be known, prefer {@link PubsubIO#readProtos(Class)}, as {@link DynamicMessage} tends to
+   * perform worse than concrete types.
+   *
+   * <p>Beam will infer a schema for the {@link DynamicMessage} schema. Note that some proto schema
+   * features are not supported by all sinks.
+   *
+   * @param domain The {@link ProtoDomain} that contains the target message and its dependencies.
+   * @param fullMessageName The full name of the message for lookup in {@code domain}.
+   */
+  @Experimental(Kind.SCHEMAS)
+  public static Read<DynamicMessage> readProtoDynamicMessage(
+      ProtoDomain domain, String fullMessageName) {
+    SerializableFunction<PubsubMessage, DynamicMessage> parser =
+        message -> {
+          try {
+            return DynamicMessage.parseFrom(
+                domain.getDescriptor(fullMessageName), message.getPayload());
+          } catch (InvalidProtocolBufferException e) {
+            throw new RuntimeException("Could not parse Pub/Sub message", e);

Review comment:
       Currently, it isn't supported, and the other methods that parse binary content for Protos or Avro throw a RuntimeException like here. 
   
   I do think it would be valuable to have something like `withDeadLetter` to configure the returned `Read` object, but I think that's for a separate PR/Jira.




-- 
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:
users@infra.apache.org