PubsubIO { @Experimental(Kind.SCHEMAS) public static Read<GenericRecord> readAvroGenericRecords(org.apache.avro.Schema avroSchema) { Schema schema = AvroUtils.getSchema(GenericRecord.class, avroSchema); AvroCoder<GenericRecord> coder = AvroCoder.of(GenericRecord.class, avroSchema); return Read.newBuilder(parsePayloadUsingCoder(coder)) .setCoder( SchemaCoder.of( schema, TypeDescriptor.of(GenericRecord.class), AvroUtils.getToRowFunction(GenericRecord.class, avroSchema), AvroUtils.getFromRowFunction(GenericRecord.class))) .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 testAvroGenericRecords() { AvroCoder<GenericRecord> coder = AvroCoder.of(GenericRecord.class, SCHEMA); List<GenericRecord> inputs = ImmutableList.of( new AvroGeneratedUser("Bob", 256, null), new AvroGeneratedUser("Alice", 128, null), new AvroGeneratedUser("Ted", null, "white")); setupTestClient(inputs, coder); PCollection<GenericRecord> read = readPipeline.apply( PubsubIO.readAvroGenericRecords(SCHEMA) .fromSubscription(SUBSCRIPTION.getPath()) .withClock(CLOCK) .withClientFactory(clientFactory)); PAssert.that(read).containsInAnyOrder(inputs); readPipeline.run(); }