Fix DruidNavigator loading multiple tables

CHANGELOG.md

@@ -500,6 +500,10 @@ Removals:
 
 ### Fixed:
 
+- [Fix the generic example for loading multiple tables](https://github.com/yahoo/fili/pull/309)
+    * Loading multiple tables caused it to hang and eventually time out.
+    * Also fixed issue causing all tables to show the same set of dimensions.
+
 - [Support for Lucene 5 indexes restored](https://github.com/yahoo/fili/pull/265)
     * Added `lucene-backward-codecs.jar` as a dependency to restore support for indexes built on earlier instances.
 

fili-generic-example/README.md

@@ -1,9 +1,15 @@
 Fili Generic Loader Application
 ==================================
 
-This application will automatically configure fili to work with **any** instance of Druid and show the basic metrics and dimensions. This lets you test what it's like using Fili without putting any effort into setting it up.
+This application will automatically configure fili to work with **any** instance
+ of Druid and show the basic metrics and dimensions. This lets you test what it's
+  like using Fili without putting any effort into setting it up.
 
-In order to set up, this will connect to druid at  [http://localhost:8081/druid/coordinator/v1](http://localhost:8081/druid/coordinator/v1). If your set up is different, you'll have to change the `bard__druid_coord` url in `applicationConfig.properties`.
+In order to set up, this will connect to druid at  [http://localhost:8081/druid/coordinator/v1](http://localhost:8081/druid/coordinator/v1).
+ If your set up is different, you'll have to change the `bard__druid_coord`,
+  `bard__non_ui_druid_broker`, `bard__ui_druid_broker` url in `applicationConfig.properties`.
+
+Note that this was last tested using [version 0.9.1](https://github.com/yahoo/fili/tree/0.9.1)
 
 ## Setup and Launching
 
@@ -15,17 +21,19 @@ In order to set up, this will connect to druid at  [http://localhost:8081/druid/
     ```
 3. Use Maven to install and launch the Fili Generic example:
 
-
-```bash
-cd fili
-mvn install
-mvn -pl fili-generic-example exec:java
-```
+    ```bash
+    cd fili
+    mvn install
+    mvn -pl fili-generic-example exec:java
+    ```
 
 - Note that if your setup is different you can adjust it by changing the default parameters below
 
     ```bash
-    mvn -pl fili-generic-example exec:java -Dbard__fili_port=9998 -Dbard__druid_coord=http://localhost:8081/druid/coordinator/v1
+    mvn -pl fili-generic-example exec:java -Dbard__fili_port=9998 \
+    -Dbard__druid_coord=http://localhost:8081/druid/coordinator/v1 \
+    -Dbard__non_ui_druid_broker=http://localhost:8082/druid/v2 \
+    -Dbard__ui_druid_broker=http://localhost:8082/druid/v2
     ```
 
 From another window, run a test query against the default druid data.
@@ -36,21 +44,21 @@ Here are some sample queries that you can run to verify your server:
 
 ### Any Server
 
-- List tables:
+- List [tables](http://localhost:9998/v1/tables):
 
       GET http://localhost:9998/v1/tables
 
-- List dimensions:  
+- List [dimensions](http://localhost:9998/v1/dimensions):  
 
       GET http://localhost:9998/v1/dimensions
 
-- List metrics:
+- List [metrics](http://localhost:9998/v1/metrics/):
 
       GET http://localhost:9998/v1/metrics/
 
 ### Specific to Wikipedia data
 
-- If everything is working, the query below
+- If everything is working, the [query below](http://localhost:9998/v1/data/wikiticker/day/?metrics=deleted&dateTime=2015-09-12/PT24H)
     ```bash
     curl "http://localhost:9998/v1/data/wikiticker/day/?metrics=deleted&dateTime=2015-09-12/PT24H" -H "Content-Type: application/json" | python -m json.tool
     ```
@@ -67,7 +75,9 @@ Here are some sample queries that you can run to verify your server:
 - Count of edits by hour for the last 72 hours:  
 
       GET http://localhost:9998/v1/data/wikiticker/day/?metrics=count&dateTime=PT72H/current
-    Note: this will should be something like the response below unless you have streaming data.
+
+    Note: this will should be something like the response below since the 
+    wikiticker table doesn't have data for the past 72 hours from now.
     ```json
     {
         "rows": [],
@@ -77,7 +87,8 @@ Here are some sample queries that you can run to verify your server:
     }
     ```  
 
-- Show debug info, including the query sent to Druid:  
+- Show [debug info](http://localhost:9998/v1/data/wikiticker/day/?format=debug&metrics=count&dateTime=PT72H/current),
+ including the query sent to Druid:  
 
       GET http://localhost:9998/v1/data/wikiticker/day/?format=debug&metrics=count&dateTime=PT72H/current
 
@@ -89,5 +100,12 @@ Here are some sample queries that you can run to verify your server:
 ## Importing and Running in IntelliJ
 
 1. In IntelliJ, go to `File -> Open`
+
 2. Select the `pom.xml` file at the root of the project
-3. Run `GenericMain` which can be found in `fili-generic-example` (e.g. right click and choose run)
+
+    **NOTE:** if you're running this locally and haven't changed any settings (like the Wikipedia example) 
+    you can **skip step 3**.
+3. Under `src/main/resources/applicationConfig.properties`, change `bard__non_ui_druid_broker`, 
+`bard__ui_druid_broker`, `bard__druid_coord`, and other properties.
+
+4. Run `GenericMain` which can be found in `fili-generic-example` (e.g. right click and choose run)

fili-generic-example/src/main/java/com/yahoo/wiki/webservice/data/config/auto/DruidNavigator.java

@@ -69,15 +69,35 @@ public List<? extends DataSourceConfiguration> get() {
      * ["wikiticker"]
      */
     private void loadAllDatasources() {
-        queryDruid(rootNode -> {
+        List<Future<Response>> fullTableResponses = new ArrayList<>();
+        Future<Response> responseFuture = queryDruid(rootNode -> {
             if (rootNode.isArray()) {
                 rootNode.forEach(jsonNode -> {
                     TableConfig tableConfig = new TableConfig(jsonNode.asText());
-                    loadTable(tableConfig);
+                    Future<Response> tableResponseFuture = loadTable(tableConfig);
+                    fullTableResponses.add(tableResponseFuture);
                     tableConfigurations.add(tableConfig);
                 });
             }
         }, COORDINATOR_TABLES_PATH);
+
+        // wait until list of table names are loaded before processing continues (i.e. ["t1","t2"])
+        try {
+            responseFuture.get(30, TimeUnit.SECONDS);
+        } catch (TimeoutException | InterruptedException | ExecutionException e) {
+            LOG.error("Interrupted while waiting for a response from druid", e);
+            throw new RuntimeException("Unable to automatically configure correctly, no response from druid.", e);
+        }
+
+        // force each individual table to finish loading (i.e. loadTable(t1) and loadTable(t2) have finished)
+        fullTableResponses.forEach(future -> {
+            try {
+                future.get(30, TimeUnit.SECONDS);
+            } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                LOG.error("Interrupted while building tables", e);
+                throw new RuntimeException("Unable to configure, couldn't fetch table data.", e);
+            }
+        });
     }
 
     /**
@@ -99,11 +119,13 @@ private void loadAllDatasources() {
      * }
      *
      * @param table The TableConfig to be loaded with queries against druid.
+     *
+     * @return future response for the query loading the table
      */
-    private void loadTable(TableConfig table) {
+    private Future<Response> loadTable(TableConfig table) {
         String url = COORDINATOR_TABLES_PATH + table.getName() + "/?full";
         String segmentsPath = "segments";
-        queryDruid(rootNode -> {
+        return queryDruid(rootNode -> {
             if (rootNode.get(segmentsPath).size() == 0) {
                 LOG.error("The segments list returned from {} was empty.", url);
                 throw new RuntimeException("Can't configure table without segment data.");
@@ -192,10 +214,12 @@ private void loadTimeGrains(TableConfig tableConfig, JsonNode segmentJson) {
      *
      * @param successCallback  The callback to be done if the query succeeds.
      * @param url  The url to send the query to.
+     *
+     * @return future response for the druid query
      */
-    private void queryDruid(SuccessCallback successCallback, String url) {
+    private Future<Response> queryDruid(SuccessCallback successCallback, String url) {
         LOG.debug("Fetching " + url);
-        Future<Response> responseFuture = druidWebService.getJsonObject(
+        return druidWebService.getJsonObject(
                 rootNode -> {
                     LOG.debug("Succesfully fetched " + url);
                     successCallback.invoke(rootNode);
@@ -210,13 +234,5 @@ private void queryDruid(SuccessCallback successCallback, String url) {
                 },
                 url
         );
-
-        try {
-            //calling get so we wait until responses are loaded before returning and processing continues
-            responseFuture.get(30, TimeUnit.SECONDS);
-        } catch (TimeoutException | InterruptedException | ExecutionException e) {
-            LOG.error("Interrupted while waiting for a response from druid", e);
-            throw new RuntimeException("Unable to automatically configure correctly, no response from druid.", e);
-        }
     }
 }

fili-generic-example/src/main/java/com/yahoo/wiki/webservice/data/config/dimension/GenericDimensionConfigs.java

@@ -17,39 +17,45 @@
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.Map;
+import java.util.HashMap;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 /**
  * Hold all the dimension configurations for a generic druid configuration.
  */
 public class GenericDimensionConfigs {
-    private final Set<DimensionConfig> dimensionConfigs;
+    private final Map<String, Set<DimensionConfig>> dataSourceToDimensionConfigs;
 
     /**
      * Construct the dimension configurations.
      *
      * @param configLoader  Supplies DataSourceConfigurations to build the dimensions from.
      */
     public GenericDimensionConfigs(Supplier<List<? extends DataSourceConfiguration>> configLoader) {
-        dimensionConfigs = configLoader.get().stream()
-                .flatMap(tableName -> tableName.getDimensions().stream())
-                .map(dimensionName -> new DefaultKeyValueStoreDimensionConfig(
-                        () -> dimensionName,
-                        dimensionName,
-                        "",
-                        dimensionName,
-                        "General",
-                        getDefaultFields(),
-                        getDefaultKeyValueStore(dimensionName),
-                        getDefaultSearchProvider(dimensionName)
-                ))
-                .collect(
-                        Collectors.collectingAndThen(
-                                Collectors.toSet(),
-                                Collections::unmodifiableSet
-                        )
-                );
+        dataSourceToDimensionConfigs = new HashMap<>();
+        configLoader.get()
+                .forEach(dataSourceConfiguration -> {
+                    Set<DimensionConfig> tableDimensionConfigs = dataSourceConfiguration.getDimensions().stream()
+                            .map(dimensionName -> new DefaultKeyValueStoreDimensionConfig(
+                                            () -> dimensionName,
+                                            dimensionName,
+                                            "",
+                                            dimensionName,
+                                            "General",
+                                            getDefaultFields(),
+                                            getDefaultKeyValueStore(dimensionName),
+                                            getDefaultSearchProvider(dimensionName)
+                                    )
+                            ).collect(
+                                    Collectors.collectingAndThen(
+                                            Collectors.toSet(),
+                                            Collections::unmodifiableSet
+                                    ));
+
+                    dataSourceToDimensionConfigs.put(dataSourceConfiguration.getName(), tableDimensionConfigs);
+                });
     }
 
     /**
@@ -58,13 +64,27 @@ public GenericDimensionConfigs(Supplier<List<? extends DataSourceConfiguration>>
      * @return set of dimension configurations
      */
     public Set<DimensionConfig> getAllDimensionConfigurations() {
-        return dimensionConfigs;
+        return dataSourceToDimensionConfigs.values().stream()
+                .flatMap(Set::stream)
+                .distinct()
+                .collect(Collectors.toSet());
+    }
+
+    /**
+     * Get all the dimension configurations associated with this datasource.
+     *
+     * @param dataSourceConfiguration  The datasource configuration's dimensions to load
+     *
+     * @return the dimension configurations for this datasource
+     */
+    public Set<DimensionConfig> getDimensionConfigs(DataSourceConfiguration dataSourceConfiguration) {
+        return dataSourceToDimensionConfigs.getOrDefault(dataSourceConfiguration.getName(), Collections.emptySet());
     }
 
     /**
      * Lazily provide a KeyValueStore for this store name.
      *
-     * @param storeName  the name for the key value store
+     * @param storeName  The name for the key value store
      *
      * @return A KeyValueStore instance
      */

fili-generic-example/src/main/java/com/yahoo/wiki/webservice/data/config/metric/GenericMetricLoader.java

@@ -40,7 +40,7 @@ public GenericMetricLoader(
     }
 
     /**
-     * (Re)Initialize the metric makers with the given metric dictionary.
+     * Initialize the metric makers with the given metric dictionary.
      *
      * @param metricDictionary  Metric dictionary to use for generating the metric makers.
      */