Merge pull request apache#2382 from himanshug/broker_segment_tier_sel…

…ection

at broker, if configured, only add segments from specific tiers to the timeline

docs/content/configuration/broker.md

@@ -95,3 +95,10 @@ You can optionally only configure caching to be enabled on the broker by setting
 |`druid.broker.cache.cacheBulkMergeLimit`|positive integer or 0|Queries with more segments than this number will not attempt to fetch from cache at the broker level, leaving potential caching fetches (and cache result merging) to the historicals|`Integer.MAX_VALUE`|
 
 See [cache configuration](caching.html) for how to configure cache settings.
+
+### Others
+
+|Property|Possible Values|Description|Default|
+|--------|---------------|-----------|-------|
+|`druid.broker.segment.watchedTiers`|List of strings|Broker watches the segment announcements from nodes serving segments to build cache of which node is serving which segments, this configuration allows to only consider segments being served from a whitelist of tiers. By default, Broker would consider all. This can be used to partition your dataSources in specific historical tiers and configure brokers in partitions so that they are only queryable for specific dataSources.|none|
+

server/src/main/java/io/druid/client/BrokerSegmentWatcherConfig.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.client;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Set;
+
+/**
+ */
+public class BrokerSegmentWatcherConfig
+{
+  @JsonProperty
+  private Set<String> watchedTiers = null;
+
+  public Set<String> getWatchedTiers()
+  {
+    return watchedTiers;
+  }
+}

server/src/main/java/io/druid/client/BrokerServerView.java

@@ -67,6 +67,7 @@ public class BrokerServerView implements TimelineServerView
   private final ServerInventoryView baseView;
   private final TierSelectorStrategy tierSelectorStrategy;
   private final ServiceEmitter emitter;
+  private final BrokerSegmentWatcherConfig segmentWatcherConfig;
 
   private volatile boolean initialized = false;
 
@@ -78,7 +79,8 @@ public BrokerServerView(
       @Client HttpClient httpClient,
       ServerInventoryView baseView,
       TierSelectorStrategy tierSelectorStrategy,
-      ServiceEmitter emitter
+      ServiceEmitter emitter,
+      BrokerSegmentWatcherConfig segmentWatcherConfig
   )
   {
     this.warehouse = warehouse;
@@ -88,6 +90,7 @@ public BrokerServerView(
     this.baseView = baseView;
     this.tierSelectorStrategy = tierSelectorStrategy;
     this.emitter = emitter;
+    this.segmentWatcherConfig = segmentWatcherConfig;
 
     this.clients = Maps.newConcurrentMap();
     this.selectors = Maps.newHashMap();
@@ -188,6 +191,11 @@ private QueryableDruidServer removeServer(DruidServer server)
 
   private void serverAddedSegment(final DruidServerMetadata server, final DataSegment segment)
   {
+    if (segmentWatcherConfig.getWatchedTiers() != null && !segmentWatcherConfig.getWatchedTiers()
+                                                                               .contains(server.getTier())) {
+      return;
+    }
+
     String segmentId = segment.getIdentifier();
     synchronized (lock) {
       log.debug("Adding segment[%s] for server[%s]", segment, server);
@@ -216,6 +224,11 @@ private void serverAddedSegment(final DruidServerMetadata server, final DataSegm
 
   private void serverRemovedSegment(DruidServerMetadata server, DataSegment segment)
   {
+    if (segmentWatcherConfig.getWatchedTiers() != null && !segmentWatcherConfig.getWatchedTiers()
+                                                                               .contains(server.getTier())) {
+      return;
+    }
+
     String segmentId = segment.getIdentifier();
     final ServerSelector selector;
 

server/src/test/java/io/druid/client/BrokerSegmentWatcherConfigTest.java

@@ -0,0 +1,61 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.client;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableSet;
+import io.druid.segment.TestHelper;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ */
+public class BrokerSegmentWatcherConfigTest
+{
+  private static final ObjectMapper MAPPER = TestHelper.getObjectMapper();
+
+  @Test
+  public void testSerde() throws Exception
+  {
+    //defaults
+    String json = "{}";
+
+    BrokerSegmentWatcherConfig config = MAPPER.readValue(
+        MAPPER.writeValueAsString(
+            MAPPER.readValue(json, BrokerSegmentWatcherConfig.class)
+        ),
+        BrokerSegmentWatcherConfig.class
+    );
+
+    Assert.assertNull(config.getWatchedTiers());
+
+    //non-defaults
+    json = "{ \"watchedTiers\": [\"t1\", \"t2\"] }";
+
+    config = MAPPER.readValue(
+        MAPPER.writeValueAsString(
+            MAPPER.readValue(json, BrokerSegmentWatcherConfig.class)
+        ),
+        BrokerSegmentWatcherConfig.class
+    );
+
+    Assert.assertEquals(ImmutableSet.of("t1", "t2"), config.getWatchedTiers());
+  }
+}

server/src/test/java/io/druid/client/BrokerServerViewTest.java

@@ -333,7 +333,8 @@ public CallbackAction segmentViewInitialized()
         EasyMock.createMock(HttpClient.class),
         baseView,
         new HighestPriorityTierSelectorStrategy(new RandomServerSelectorStrategy()),
-        new NoopServiceEmitter()
+        new NoopServiceEmitter(),
+        new BrokerSegmentWatcherConfig()
     );
 
     baseView.start();

services/src/main/java/io/druid/cli/CliBroker.java

@@ -25,6 +25,7 @@
 import com.google.inject.name.Names;
 import com.metamx.common.logger.Logger;
 import io.airlift.airline.Command;
+import io.druid.client.BrokerSegmentWatcherConfig;
 import io.druid.client.BrokerServerView;
 import io.druid.client.CachingClusteredClient;
 import io.druid.client.TimelineServerView;
@@ -98,6 +99,7 @@ public void configure(Binder binder)
             JsonConfigProvider.bind(binder, "druid.broker.select.tier.custom", CustomTierSelectorStrategyConfig.class);
             JsonConfigProvider.bind(binder, "druid.broker.balancer", ServerSelectorStrategy.class);
             JsonConfigProvider.bind(binder, "druid.broker.retryPolicy", RetryQueryRunnerConfig.class);
+            JsonConfigProvider.bind(binder, "druid.broker.segment", BrokerSegmentWatcherConfig.class);
 
             binder.bind(QuerySegmentWalker.class).to(ClientQuerySegmentWalker.class).in(LazySingleton.class);