@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 testInvalidConfigInvalidFormat() {
    PCollectionRowTuple begin = PCollectionRowTuple.empty(p);
    assertThrows(
        IllegalArgumentException.class,
        () ->
            begin.apply(
                new PubsubReadSchemaTransformProvider()
                    .from(
                        PubsubReadSchemaTransformConfiguration.builder()
                            .setSchema(SCHEMA)
                            .setFormat("BadFormat")
                            .setSubscription(SUBSCRIPTION)
                            .build())));
    p.run().waitUntilFinish();
  }