@Override
  public BeamSqlTable buildBeamSqlTable(Table table) {
    Schema schema = table.getSchema();
    ObjectNode properties = table.getProperties();

    Optional<ParsedLocation> parsedLocation = Optional.empty();
    if (!Strings.isNullOrEmpty(table.getLocation())) {
      parsedLocation = Optional.of(parseLocation(checkArgumentNotNull(table.getLocation())));
    }
    List<String> topics =
        mergeParam(parsedLocation.map(loc -> loc.topic), (ArrayNode) properties.get("topics"));
    List<String> allBootstrapServers =
        mergeParam(
            parsedLocation.map(loc -> loc.brokerLocation),
            (ArrayNode) properties.get("bootstrap_servers"));
    String bootstrapServers = String.join(",", allBootstrapServers);

    Optional<String> payloadFormat =
        properties.has("format")
            ? Optional.of(properties.get("format").asText())
            : Optional.empty();
    if (Schemas.isNestedSchema(schema)) {
      Optional<PayloadSerializer> serializer =
          payloadFormat.map(
              format ->
                  PayloadSerializers.getSerializer(
                      format,
                      checkArgumentNotNull(schema.getField(PAYLOAD_FIELD).getType().getRowSchema()),
                      TableUtils.convertNode2Map(properties)));
      return new NestedPayloadKafkaTable(schema, bootstrapServers, topics, serializer);
    } else {
      /*
       * CSV is handled separately because multiple rows can be produced from a single message, which
       * adds complexity to payload extraction. It remains here and as the default because it is the
       * historical default, but it will not be extended to support attaching extended attributes to
       * rows.
       */
      if (payloadFormat.orElse("csv").equals("csv")) {
        return new BeamKafkaCSVTable(schema, bootstrapServers, topics);
      }
      PayloadSerializer serializer =
          PayloadSerializers.getSerializer(
              payloadFormat.get(), schema, TableUtils.convertNode2Map(properties));
      return new PayloadSerializerKafkaTable(schema, bootstrapServers, topics, serializer);
    }
  }
@Test
  public void testBuildWithExtraTopics() {
    Table table = mockTableWithExtraTopics("hello", ImmutableList.of("topic2", "topic3"));
    BeamSqlTable sqlTable = provider.buildBeamSqlTable(table);

    assertNotNull(sqlTable);
    assertTrue(sqlTable instanceof BeamKafkaCSVTable);

    BeamKafkaCSVTable kafkaTable = (BeamKafkaCSVTable) sqlTable;
    assertEquals(LOCATION_BROKER, kafkaTable.getBootstrapServers());
    assertEquals(ImmutableList.of(LOCATION_TOPIC, "topic2", "topic3"), kafkaTable.getTopics());
  }