PubsubIO { @Experimental(Kind.SCHEMAS) public static <T> Read<T> readAvrosWithBeamSchema(Class<T> clazz) { if (clazz.equals(GenericRecord.class)) { throw new IllegalArgumentException("For GenericRecord, please call readAvroGenericRecords"); } org.apache.avro.Schema avroSchema = ReflectData.get().getSchema(clazz); AvroCoder<T> coder = AvroCoder.of(clazz); Schema schema = AvroUtils.getSchema(clazz, null); return Read.newBuilder(parsePayloadUsingCoder(coder)) .setCoder( SchemaCoder.of( schema, TypeDescriptor.of(clazz), AvroUtils.getToRowFunction(clazz, avroSchema), AvroUtils.getFromRowFunction(clazz))) .build(); } private  PubsubIO(); static Read<PubsubMessage> readMessages(); static Read<PubsubMessage> readMessagesWithMessageId(); static Read<PubsubMessage> readMessagesWithAttributes(); static Read<PubsubMessage> readMessagesWithAttributesAndMessageId(); static Read<String> readStrings(); static Read<T> readProtos(Class<T> messageClass); static Read<T> readAvros(Class<T> clazz); static Read<T> readMessagesWithCoderAndParseFn(
      Coder<T> coder, SimpleFunction<PubsubMessage, T> parseFn); @Experimental(Kind.SCHEMAS) static Read<GenericRecord> readAvroGenericRecords(org.apache.avro.Schema avroSchema); @Experimental(Kind.SCHEMAS) static Read<T> readAvrosWithBeamSchema(Class<T> clazz); static Write<PubsubMessage> writeMessages(); static Write<String> writeStrings(); static Write<T> writeProtos(Class<T> messageClass); static Write<T> writeAvros(Class<T> clazz);  }
@Test public void testAvroPojo() { AvroCoder<GenericClass> coder = AvroCoder.of(GenericClass.class); List<GenericClass> inputs = Lists.newArrayList( new GenericClass( 1, "foo", new DateTime().withDate(2019, 10, 1).withZone(DateTimeZone.UTC)), new GenericClass( 2, "bar", new DateTime().withDate(1986, 10, 1).withZone(DateTimeZone.UTC))); setupTestClient(inputs, coder); PCollection<GenericClass> read = readPipeline.apply( PubsubIO.readAvrosWithBeamSchema(GenericClass.class) .fromSubscription(SUBSCRIPTION.getPath()) .withClock(CLOCK) .withClientFactory(clientFactory)); PAssert.that(read).containsInAnyOrder(inputs); readPipeline.run(); }
@Test public void testAvroSpecificRecord() { AvroCoder<AvroGeneratedUser> coder = AvroCoder.of(AvroGeneratedUser.class); List<AvroGeneratedUser> inputs = ImmutableList.of( new AvroGeneratedUser("Bob", 256, null), new AvroGeneratedUser("Alice", 128, null), new AvroGeneratedUser("Ted", null, "white")); setupTestClient(inputs, coder); PCollection<AvroGeneratedUser> read = readPipeline.apply( PubsubIO.readAvrosWithBeamSchema(AvroGeneratedUser.class) .fromSubscription(SUBSCRIPTION.getPath()) .withClock(CLOCK) .withClientFactory(clientFactory)); PAssert.that(read).containsInAnyOrder(inputs); readPipeline.run(); }