static Schema fromPubsubSchema(com.google.api.services.pubsub.model.Schema pubsubSchema) {
    if (!schemaTypeToConversionFnMap.containsKey(pubsubSchema.getType())) {
      throw new IllegalArgumentException(
          String.format(
              "Pub/Sub schema type %s is not supported at this time", pubsubSchema.getType()));
    }
    SerializableFunction<String, Schema> definitionToSchemaFn =
        schemaTypeToConversionFnMap.get(pubsubSchema.getType());
    return definitionToSchemaFn.apply(pubsubSchema.getDefinition());
  }
@Test
  public void fromPubsubSchema() {
    assertThrows(
        "null definition should throw an exception",
        NullPointerException.class,
        () ->
            PubsubClient.fromPubsubSchema(
                new com.google.api.services.pubsub.model.Schema().setType("AVRO")));

    assertThrows(
        "null definition should throw an exception",
        SchemaParseException.class,
        () ->
            PubsubClient.fromPubsubSchema(
                com.google.pubsub.v1.Schema.newBuilder().setType(Schema.Type.AVRO).build()));

    String badSchema =
        "{\"type\": \"record\", \"name\": \"Avro\",\"fields\": [{\"name\": \"bad\", \"type\": \"notatype\"}]}";
    String goodSchema =
        "{"
            + " \"type\" : \"record\","
            + " \"name\" : \"Avro\","
            + " \"fields\" : ["
            + "   {"
            + "     \"name\" : \"StringField\","
            + "     \"type\" : \"string\""
            + "   },"
            + "   {"
            + "     \"name\" : \"FloatField\","
            + "     \"type\" : \"float\""
            + "   },"
            + "   {"
            + "     \"name\" : \"IntField\","
            + "     \"type\" : \"int\""
            + "   },"
            + "   {"
            + "     \"name\" : \"LongField\","
            + "     \"type\" : \"long\""
            + "   },"
            + "   {"
            + "     \"name\" : \"DoubleField\","
            + "     \"type\" : \"double\""
            + "   },"
            + "   {"
            + "     \"name\" : \"BytesField\","
            + "     \"type\" : \"bytes\""
            + "   },"
            + "   {"
            + "     \"name\" : \"BooleanField\","
            + "     \"type\" : \"boolean\""
            + "   }"
            + " ]"
            + "}";

    assertThrows(
        "unsupported Schema type should throw an exception",
        IllegalArgumentException.class,
        () ->
            PubsubClient.fromPubsubSchema(
                new com.google.api.services.pubsub.model.Schema()
                    .setType("PROTOCOL_BUFFER")
                    .setDefinition(goodSchema)));

    assertThrows(
        "'notatype' Avro type should throw an exception",
        SchemaParseException.class,
        () ->
            PubsubClient.fromPubsubSchema(
                new com.google.api.services.pubsub.model.Schema()
                    .setType("AVRO")
                    .setDefinition(badSchema)));

    assertEquals(
        org.apache.beam.sdk.schemas.Schema.of(
            org.apache.beam.sdk.schemas.Schema.Field.of(
                "StringField", org.apache.beam.sdk.schemas.Schema.FieldType.STRING),
            org.apache.beam.sdk.schemas.Schema.Field.of(
                "FloatField", org.apache.beam.sdk.schemas.Schema.FieldType.FLOAT),
            org.apache.beam.sdk.schemas.Schema.Field.of(
                "IntField", org.apache.beam.sdk.schemas.Schema.FieldType.INT32),
            org.apache.beam.sdk.schemas.Schema.Field.of(
                "LongField", org.apache.beam.sdk.schemas.Schema.FieldType.INT64),
            org.apache.beam.sdk.schemas.Schema.Field.of(
                "DoubleField", org.apache.beam.sdk.schemas.Schema.FieldType.DOUBLE),
            org.apache.beam.sdk.schemas.Schema.Field.of(
                "BytesField", org.apache.beam.sdk.schemas.Schema.FieldType.BYTES),
            org.apache.beam.sdk.schemas.Schema.Field.of(
                "BooleanField", org.apache.beam.sdk.schemas.Schema.FieldType.BOOLEAN)),
        PubsubClient.fromPubsubSchema(
            new com.google.api.services.pubsub.model.Schema()
                .setType("AVRO")
                .setDefinition(goodSchema)));
  }