public static <T> Read<T> readAvrosWithBeamSchema(Class<T> clazz) {
    if (clazz.equals(GenericRecord.class)) {
      throw new IllegalArgumentException("For GenericRecord, please call readAvroGenericRecords");
    }
    AvroCoder<T> coder = AvroCoder.of(clazz);
    org.apache.avro.Schema avroSchema = coder.getSchema();
    Schema schema = AvroUtils.getSchema(clazz, avroSchema);
    return Read.newBuilder(parsePayloadUsingCoder(coder))
        .setCoder(
            SchemaCoder.of(
                schema,
                TypeDescriptor.of(clazz),
                AvroUtils.getToRowFunction(clazz, avroSchema),
                AvroUtils.getFromRowFunction(clazz)))
        .build();
  }
@Test
  public void testAvroSpecificRecord() {
    AvroCoder<AvroGeneratedUser> coder = AvroCoder.specific(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 =
        pipeline.apply(
            PubsubIO.readAvrosWithBeamSchema(AvroGeneratedUser.class)
                .fromSubscription(SUBSCRIPTION.getPath())
                .withClock(CLOCK)
                .withClientFactory(clientFactory));
    PAssert.that(read).containsInAnyOrder(inputs);
    pipeline.run();
  }