class PubsubJsonTableProvider extends InMemoryMetaTableProvider {

  @Override
  public BeamSqlTable buildBeamSqlTable(Table tableDefintion) {
    JSONObject tableProperties = tableDefintion.getProperties();
    String timestampAttributeKey = tableProperties.getString("timestampAttributeKey");
    String deadLetterQueue = tableProperties.getString("deadLetterQueue");
    validateDlq(deadLetterQueue);

    Schema schema = tableDefintion.getSchema();
    validateEventTimestamp(schema);

    PubsubIOTableConfiguration config =
        PubsubIOTableConfiguration.builder()
            .setSchema(schema)
            .setTimestampAttribute(timestampAttributeKey)
            .setDeadLetterQueue(deadLetterQueue)
            .setTopic(tableDefintion.getLocation())
            .setUseFlatSchema(!definesAttributeAndPayload(schema))
            .build();

    return PubsubIOJsonTable.withConfiguration(config);
  }

}