Merge pull request apache#251 from metamx/fix-config

Make dynamic master resource configuration work again

indexing-hadoop/src/main/java/io/druid/indexer/HadoopDruidIndexerConfig.java

@@ -35,7 +35,6 @@
 import com.google.common.collect.Lists;
 import com.metamx.common.Granularity;
 import com.metamx.common.ISE;
-import com.metamx.common.MapUtils;
 import com.metamx.common.guava.FunctionalIterable;
 import com.metamx.common.logger.Logger;
 import io.druid.common.utils.JodaUtils;
@@ -52,8 +51,6 @@
 import io.druid.indexer.updater.DbUpdaterJobSpec;
 import io.druid.jackson.DefaultObjectMapper;
 import io.druid.query.aggregation.AggregatorFactory;
-import io.druid.segment.serde.Registererer;
-import io.druid.segment.serde.Registererers;
 import io.druid.timeline.DataSegment;
 import io.druid.timeline.partition.ShardSpec;
 import org.apache.hadoop.conf.Configuration;
@@ -65,11 +62,9 @@
 import org.joda.time.Interval;
 import org.joda.time.format.ISODateTimeFormat;
 
-import javax.annotation.Nullable;
 import java.io.File;
 import java.io.IOException;
 import java.nio.charset.Charset;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;

server/src/main/java/io/druid/server/http/MasterSegmentSettingsResource.java → server/src/main/java/io/druid/server/http/MasterDynamicConfigsResource.java

@@ -20,7 +20,7 @@
 package io.druid.server.http;
 
 import io.druid.common.config.JacksonConfigManager;
-import io.druid.server.master.MasterSegmentSettings;
+import io.druid.server.master.MasterDynamicConfig;
 
 import javax.inject.Inject;
 import javax.ws.rs.Consumes;
@@ -33,38 +33,38 @@
 /**
  */
 @Path("/master/config")
-public class MasterSegmentSettingsResource
+public class MasterDynamicConfigsResource
 {
   private final JacksonConfigManager manager;
 
   @Inject
-  public MasterSegmentSettingsResource(
+  public MasterDynamicConfigsResource(
       JacksonConfigManager manager
   )
   {
-    this.manager=manager;
+    this.manager = manager;
   }
+
   @GET
   @Produces("application/json")
   public Response getDynamicConfigs()
   {
-    Response.ResponseBuilder builder = Response.status(Response.Status.OK)
-                                               .entity(
-                                                   manager.watch(MasterSegmentSettings.CONFIG_KEY,MasterSegmentSettings.class).get()
-                                               );
-    return builder.build();
+    return Response.ok(
+        manager.watch(
+            MasterDynamicConfig.CONFIG_KEY,
+            MasterDynamicConfig.class
+        ).get()
+    ).build();
   }
 
   @POST
   @Consumes("application/json")
-  public Response setDynamicConfigs(
-      final MasterSegmentSettings masterSegmentSettings
-  )
+  public Response setDynamicConfigs(final MasterDynamicConfig dynamicConfig)
   {
-    if (!manager.set(MasterSegmentSettings.CONFIG_KEY, masterSegmentSettings)) {
+    if (!manager.set(MasterDynamicConfig.CONFIG_KEY, dynamicConfig)) {
       return Response.status(Response.Status.BAD_REQUEST).build();
     }
-    return Response.status(Response.Status.OK).build();
+    return Response.ok().build();
   }
 
 }

server/src/main/java/io/druid/server/http/MasterResource.java

@@ -22,6 +22,7 @@
 import com.google.inject.Inject;
 import io.druid.server.master.DruidMaster;
 import io.druid.server.master.LoadPeonCallback;
+import io.druid.server.master.MasterDynamicConfig;
 
 import javax.ws.rs.Consumes;
 import javax.ws.rs.GET;
@@ -30,7 +31,6 @@
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.Response;
 import java.util.List;
-import java.util.Map;
 
 /**
  */
@@ -113,8 +113,8 @@ protected void execute()
   @GET
   @Path("/loadstatus")
   @Produces("application/json")
-  public Map<String, Double> getLoadStatus()
+  public Response getLoadStatus()
   {
-    return master.getLoadStatus();
+    return Response.ok(master.getLoadStatus()).build();
   }
 }

server/src/main/java/io/druid/server/master/DruidMaster.java

@@ -84,6 +84,7 @@ public class DruidMaster
 
   private volatile boolean started = false;
   private volatile boolean master = false;
+  private volatile AtomicReference<MasterDynamicConfig> dynamicConfigs;
 
   private final DruidMasterConfig config;
   private final ZkPathsConfig zkPaths;
@@ -98,7 +99,6 @@ public class DruidMaster
   private final LoadQueueTaskMaster taskMaster;
   private final Map<String, LoadQueuePeon> loadManagementPeons;
   private final AtomicReference<LeaderLatch> leaderLatch;
-  private volatile AtomicReference<MasterSegmentSettings> segmentSettingsAtomicReference;
 
   @Inject
   public DruidMaster(
@@ -161,7 +161,7 @@ public DruidMaster(
     this.exec = scheduledExecutorFactory.create(1, "Master-Exec--%d");
 
     this.leaderLatch = new AtomicReference<>(null);
-    this.segmentSettingsAtomicReference = new AtomicReference<>(null);
+    this.dynamicConfigs = new AtomicReference<>(null);
     this.loadManagementPeons = loadQueuePeonMap;
   }
 
@@ -215,6 +215,11 @@ public Map<String, Double> getLoadStatus()
     return loadStatus;
   }
 
+  public MasterDynamicConfig getDynamicConfigs()
+  {
+    return dynamicConfigs.get();
+  }
+
   public void removeSegment(DataSegment segment)
   {
     log.info("Removing Segment[%s]", segment);
@@ -471,10 +476,10 @@ private void becomeMaster()
         serverInventoryView.start();
 
         final List<Pair<? extends MasterRunnable, Duration>> masterRunnables = Lists.newArrayList();
-        segmentSettingsAtomicReference = configManager.watch(
-            MasterSegmentSettings.CONFIG_KEY,
-            MasterSegmentSettings.class,
-            new MasterSegmentSettings.Builder().build()
+        dynamicConfigs = configManager.watch(
+            MasterDynamicConfig.CONFIG_KEY,
+            MasterDynamicConfig.class,
+            new MasterDynamicConfig.Builder().build()
         );
         masterRunnables.add(Pair.of(new MasterComputeManagerRunnable(), config.getMasterPeriod()));
         if (indexingServiceClient != null) {
@@ -663,7 +668,7 @@ public void run()
             DruidMasterRuntimeParams.newBuilder()
                                     .withStartTime(startTime)
                                     .withDatasources(databaseSegmentManager.getInventory())
-                                    .withMasterSegmentSettings(segmentSettingsAtomicReference.get())
+                                    .withDynamicConfigs(dynamicConfigs.get())
                                     .withEmitter(emitter)
                                     .build();
 
@@ -750,13 +755,11 @@ public boolean apply(
                                .withLoadManagementPeons(loadManagementPeons)
                                .withSegmentReplicantLookup(segmentReplicantLookup)
                                .withBalancerReferenceTimestamp(DateTime.now())
-                               .withMasterSegmentSettings(segmentSettingsAtomicReference.get())
+                               .withDynamicConfigs(dynamicConfigs.get())
                                .build();
                 }
               },
-              new DruidMasterRuleRunner(
-                  DruidMaster.this, config.getReplicantLifetime(), config.getReplicantThrottleLimit()
-              ),
+              new DruidMasterRuleRunner(DruidMaster.this),
               new DruidMasterCleanup(DruidMaster.this),
               new DruidMasterBalancer(DruidMaster.this),
               new DruidMasterLogger()

server/src/main/java/io/druid/server/master/DruidMasterBalancer.java

@@ -79,7 +79,7 @@ public DruidMasterRuntimeParams run(DruidMasterRuntimeParams params)
     final MasterStats stats = new MasterStats();
     final DateTime referenceTimestamp = params.getBalancerReferenceTimestamp();
     final BalancerStrategy strategy = params.getBalancerStrategyFactory().createBalancerStrategy(referenceTimestamp);
-    final int maxSegmentsToMove = params.getMasterSegmentSettings().getMaxSegmentsToMove();
+    final int maxSegmentsToMove = params.getMasterDynamicConfig().getMaxSegmentsToMove();
 
     for (Map.Entry<String, MinMaxPriorityQueue<ServerHolder>> entry :
         params.getDruidCluster().getCluster().entrySet()) {
@@ -124,7 +124,7 @@ public DruidMasterRuntimeParams run(DruidMasterRuntimeParams params)
         }
       }
       stats.addToTieredStat("movedCount", tier, currentlyMovingSegments.get(tier).size());
-      if (params.getMasterSegmentSettings().isEmitBalancingStats()) {
+      if (params.getMasterDynamicConfig().emitBalancingStats()) {
         strategy.emitStats(tier, stats, serverHolderList);
 
       }

server/src/main/java/io/druid/server/master/DruidMasterConfig.java

@@ -54,14 +54,6 @@ public boolean isConvertSegments()
     return false;
   }
 
-  @Config("druid.master.replicant.lifetime")
-  @Default("15")
-  public abstract int getReplicantLifetime();
-
-  @Config("druid.master.replicant.throttleLimit")
-  @Default("10")
-  public abstract int getReplicantThrottleLimit();
-
   @Config("druid.master.load.timeout")
   public Duration getLoadTimeoutDelay()
   {

server/src/main/java/io/druid/server/master/DruidMasterRuleRunner.java

@@ -37,10 +37,21 @@ public class DruidMasterRuleRunner implements DruidMasterHelper
 
   private final DruidMaster master;
 
-  public DruidMasterRuleRunner(DruidMaster master, int replicantLifeTime, int replicantThrottleLimit)
+  public DruidMasterRuleRunner(DruidMaster master)
   {
+    this(
+        new ReplicationThrottler(
+            master.getDynamicConfigs().getReplicationThrottleLimit(),
+            master.getDynamicConfigs().getReplicantLifetime()
+        ),
+        master
+    );
+  }
+
+  public DruidMasterRuleRunner(ReplicationThrottler replicatorThrottler, DruidMaster master)
+  {
+    this.replicatorThrottler = replicatorThrottler;
     this.master = master;
-    this.replicatorThrottler = new ReplicationThrottler(replicantThrottleLimit, replicantLifeTime);
   }
 
   @Override