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:
[email protected]
