class KafkaTableProvider extends InMemoryMetaTableProvider {

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

    JSONObject properties = table.getProperties();
    String bootstrapServers = properties.getString("bootstrap.servers");
    JSONArray topicsArr = properties.getJSONArray("topics");
    List<String> topics = new ArrayList<>(topicsArr.size());
    for (Object topic : topicsArr) {
      topics.add(topic.toString());
    }
    return new BeamKafkaCSVTable(schema, bootstrapServers, topics);
  }

}