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); }  @Override BeamSqlTable buildBeamSqlTable(Table table); @Override String getTableType();  }
@Test public void testBuildBeamSqlTable() throws Exception { Table table = mockTable("hello"); BeamSqlTable sqlTable = provider.buildBeamSqlTable(table); assertNotNull(sqlTable); assertTrue(sqlTable instanceof BeamKafkaCSVTable); BeamKafkaCSVTable csvTable = (BeamKafkaCSVTable) sqlTable; assertEquals("localhost:9092", csvTable.getBootstrapServers()); assertEquals(ImmutableList.of("topic1", "topic2"), csvTable.getTopics()); }