public static Read<GenericRecord> readAvroGenericRecords(org.apache.avro.Schema avroSchema) {
    AvroCoder<GenericRecord> coder = AvroCoder.of(avroSchema);
    Schema schema = AvroUtils.getSchema(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();
  }
@Test
  public void testAvroGenericRecords() {
    AvroCoder<GenericRecord> coder = AvroCoder.of(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 =
        pipeline.apply(
            PubsubIO.readAvroGenericRecords(SCHEMA)
                .fromSubscription(SUBSCRIPTION.getPath())
                .withClock(CLOCK)
                .withClientFactory(clientFactory));
    PAssert.that(read).containsInAnyOrder(inputs);
    pipeline.run();
  }