@Override
  public SchemaTransform from(PubsubReadSchemaTransformConfiguration configuration) {
    if (configuration.getSubscription() == null && configuration.getTopic() == null) {
      throw new IllegalArgumentException(
          "To read from Pubsub, a subscription name or a topic name must be provided");
    }

    if (configuration.getSubscription() != null && configuration.getTopic() != null) {
      throw new IllegalArgumentException(
          "To read from Pubsub, a subscription name or a topic name must be provided. Not both.");
    }

    if (!"RAW".equals(configuration.getFormat())) {
      if ((Strings.isNullOrEmpty(configuration.getSchema())
              && !Strings.isNullOrEmpty(configuration.getFormat()))
          || (!Strings.isNullOrEmpty(configuration.getSchema())
              && Strings.isNullOrEmpty(configuration.getFormat()))) {
        throw new IllegalArgumentException(
            "A schema was provided without a data format (or viceversa). Please provide "
                + "both of these parameters to read from Pubsub, or if you would like to use the Pubsub schema service,"
                + " please leave both of these blank.");
      }
    }

    Schema payloadSchema;
    SerializableFunction<byte[], Row> payloadMapper;

    String format =
        configuration.getFormat() == null ? null : configuration.getFormat().toUpperCase();
    if ("RAW".equals(format)) {
      payloadSchema = Schema.of(Schema.Field.of("payload", Schema.FieldType.BYTES));
      payloadMapper = input -> Row.withSchema(payloadSchema).addValue(input).build();
    } else if ("JSON".equals(format)) {
      payloadSchema = JsonUtils.beamSchemaFromJsonSchema(configuration.getSchema());
      payloadMapper = JsonUtils.getJsonBytesToRowFunction(payloadSchema);
    } else if ("AVRO".equals(format)) {
      payloadSchema =
          AvroUtils.toBeamSchema(
              new org.apache.avro.Schema.Parser().parse(configuration.getSchema()));
      payloadMapper = AvroUtils.getAvroBytesToRowFunction(payloadSchema);
    } else {
      throw new IllegalArgumentException(
          String.format(
              "Format %s not supported. Only supported formats are %s",
              configuration.getFormat(), VALID_FORMATS_STR));
    }

    PubsubReadSchemaTransform transform =
        new PubsubReadSchemaTransform(configuration, payloadSchema, payloadMapper);

    if (configuration.getClientFactory() != null) {
      transform.setClientFactory(configuration.getClientFactory());
    }
    if (configuration.getClock() != null) {
      transform.setClock(configuration.getClock());
    }

    return transform;
  }
@Test
  public void testReadAvroWithError() throws IOException {
    PCollectionRowTuple begin = PCollectionRowTuple.empty(p);

    try (PubsubTestClientFactory clientFactory = clientFactory(beamRowToMessageWithError())) {
      PubsubReadSchemaTransformConfiguration config =
          PubsubReadSchemaTransformConfiguration.builder()
              .setFormat("AVRO")
              .setSchema(SCHEMA)
              .setSubscription(SUBSCRIPTION)
              .setErrorHandling(
                  PubsubReadSchemaTransformConfiguration.ErrorHandling.builder()
                      .setOutput("errors")
                      .build())
              .setClientFactory(clientFactory)
              .setClock(CLOCK)
              .build();
      SchemaTransform transform = new PubsubReadSchemaTransformProvider().from(config);
      PCollectionRowTuple reads = begin.apply(transform);

      PAssert.that(reads.get("output")).empty();

      PipelineResult result = p.run();
      result.waitUntilFinish();

      MetricResults metrics = result.metrics();
      MetricQueryResults metricResults =
          metrics.queryMetrics(
              MetricsFilter.builder()
                  .addNameFilter(
                      MetricNameFilter.named(
                          PubsubReadSchemaTransformProvider.class, "PubSub-read-error-counter"))
                  .build());

      Iterable<MetricResult<Long>> counters = metricResults.getCounters();
      if (!counters.iterator().hasNext()) {
        throw new RuntimeException("no counters available ");
      }

      Long expectedCount = 3L;
      for (MetricResult<Long> count : counters) {
        assertEquals(expectedCount, count.getAttempted());
      }
    } catch (Exception e) {
      throw e;
    }
  }