InputSchemaFactory inputSchemaFactory() {
    return inputSchemaFactory(null);
  }
@Test
  public void testInputSchemaFactory() {
    assertEquals(
        Schema.of(
            Field.of(DEFAULT_ATTRIBUTES_KEY_NAME, ATTRIBUTES_FIELD_TYPE),
            Field.of(DEFAULT_EVENT_TIMESTAMP_KEY_NAME, EVENT_TIMESTAMP_FIELD_TYPE)),
        PubsubRowToMessage.builder().build().inputSchemaFactory().buildSchema());

    assertEquals(
        Schema.of(
            Field.of(DEFAULT_ATTRIBUTES_KEY_NAME, ATTRIBUTES_FIELD_TYPE),
            Field.of(DEFAULT_EVENT_TIMESTAMP_KEY_NAME, EVENT_TIMESTAMP_FIELD_TYPE),
            Field.of(DEFAULT_PAYLOAD_KEY_NAME, FieldType.BYTES)),
        PubsubRowToMessage.builder().build().inputSchemaFactory(FieldType.BYTES).buildSchema());

    assertEquals(
        Schema.of(
            Field.of(DEFAULT_ATTRIBUTES_KEY_NAME, ATTRIBUTES_FIELD_TYPE),
            Field.of(DEFAULT_EVENT_TIMESTAMP_KEY_NAME, EVENT_TIMESTAMP_FIELD_TYPE),
            Field.of(DEFAULT_PAYLOAD_KEY_NAME, FieldType.row(ALL_DATA_TYPES_SCHEMA))),
        PubsubRowToMessage.builder()
            .build()
            .inputSchemaFactory(FieldType.row(ALL_DATA_TYPES_SCHEMA))
            .buildSchema());

    String prefix = "_";
    assertEquals(
        Schema.of(
            Field.of(prefix + ATTRIBUTES_KEY_NAME, ATTRIBUTES_FIELD_TYPE),
            Field.of(prefix + EVENT_TIMESTAMP_KEY_NAME, EVENT_TIMESTAMP_FIELD_TYPE)),
        PubsubRowToMessage.builder()
            .setKeyPrefix(prefix)
            .build()
            .inputSchemaFactory()
            .buildSchema());

    Field[] userFields = ALL_DATA_TYPES_SCHEMA.getFields().toArray(new Field[0]);
    assertEquals(
        merge(
            Schema.of(
                Field.of(DEFAULT_ATTRIBUTES_KEY_NAME, ATTRIBUTES_FIELD_TYPE),
                Field.of(DEFAULT_EVENT_TIMESTAMP_KEY_NAME, EVENT_TIMESTAMP_FIELD_TYPE)),
            ALL_DATA_TYPES_SCHEMA),
        PubsubRowToMessage.builder().build().inputSchemaFactory().buildSchema(userFields));
  }