Create basePersistDirectory only when needed

extensions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaTuningConfig.java

@@ -37,20 +37,19 @@ public class KafkaTuningConfig implements TuningConfig, AppenderatorConfig
   private final int maxRowsInMemory;
   private final int maxRowsPerSegment;
   private final Period intermediatePersistPeriod;
-  private final File basePersistDirectory;
   private final int maxPendingPersists;
   private final IndexSpec indexSpec;
   private final boolean buildV9Directly;
   private final boolean reportParseExceptions;
   private final long handoffConditionTimeout;
   private final boolean resetOffsetAutomatically;
+  private File basePersistDirectory;
 
   @JsonCreator
   public KafkaTuningConfig(
       @JsonProperty("maxRowsInMemory") Integer maxRowsInMemory,
       @JsonProperty("maxRowsPerSegment") Integer maxRowsPerSegment,
       @JsonProperty("intermediatePersistPeriod") Period intermediatePersistPeriod,
-      @JsonProperty("basePersistDirectory") File basePersistDirectory,
       @JsonProperty("maxPendingPersists") Integer maxPendingPersists,
       @JsonProperty("indexSpec") IndexSpec indexSpec,
       @JsonProperty("buildV9Directly") Boolean buildV9Directly,
@@ -60,14 +59,13 @@ public KafkaTuningConfig(
   )
   {
     // Cannot be a static because default basePersistDirectory is unique per-instance
-    final RealtimeTuningConfig defaults = RealtimeTuningConfig.makeDefaultTuningConfig(basePersistDirectory);
+    final RealtimeTuningConfig defaults = new RealtimeTuningConfig.Builder().build();
 
     this.maxRowsInMemory = maxRowsInMemory == null ? defaults.getMaxRowsInMemory() : maxRowsInMemory;
     this.maxRowsPerSegment = maxRowsPerSegment == null ? DEFAULT_MAX_ROWS_PER_SEGMENT : maxRowsPerSegment;
     this.intermediatePersistPeriod = intermediatePersistPeriod == null
                                      ? defaults.getIntermediatePersistPeriod()
                                      : intermediatePersistPeriod;
-    this.basePersistDirectory = defaults.getBasePersistDirectory();
     this.maxPendingPersists = maxPendingPersists == null ? defaults.getMaxPendingPersists() : maxPendingPersists;
     this.indexSpec = indexSpec == null ? defaults.getIndexSpec() : indexSpec;
     this.buildV9Directly = buildV9Directly == null ? defaults.getBuildV9Directly() : buildV9Directly;
@@ -88,7 +86,6 @@ public static KafkaTuningConfig copyOf(KafkaTuningConfig config)
         config.maxRowsInMemory,
         config.maxRowsPerSegment,
         config.intermediatePersistPeriod,
-        config.basePersistDirectory,
         config.maxPendingPersists,
         config.indexSpec,
         config.buildV9Directly,
@@ -160,34 +157,40 @@ public boolean isResetOffsetAutomatically()
 
   public KafkaTuningConfig withBasePersistDirectory(File dir)
   {
-    return new KafkaTuningConfig(
+    KafkaTuningConfig newKafkaTuningConfig = new KafkaTuningConfig(
         maxRowsInMemory,
         maxRowsPerSegment,
         intermediatePersistPeriod,
-        dir,
         maxPendingPersists,
         indexSpec,
         buildV9Directly,
         reportParseExceptions,
         handoffConditionTimeout,
         resetOffsetAutomatically
     );
+    newKafkaTuningConfig.basePersistDirectory = dir;
+    return newKafkaTuningConfig;
   }
 
   public KafkaTuningConfig withMaxRowsInMemory(int rows)
   {
-    return new KafkaTuningConfig(
+    KafkaTuningConfig newKafkaTuningConfig = new KafkaTuningConfig(
         rows,
         maxRowsPerSegment,
         intermediatePersistPeriod,
-        basePersistDirectory,
         maxPendingPersists,
         indexSpec,
         buildV9Directly,
         reportParseExceptions,
         handoffConditionTimeout,
         resetOffsetAutomatically
     );
+
+    if (basePersistDirectory != null) {
+      return newKafkaTuningConfig.withBasePersistDirectory(basePersistDirectory);
+    }
+
+    return newKafkaTuningConfig;
   }
 
   @Override
@@ -228,11 +231,13 @@ public boolean equals(Object o)
         : that.intermediatePersistPeriod != null) {
       return false;
     }
+
     if (basePersistDirectory != null
         ? !basePersistDirectory.equals(that.basePersistDirectory)
         : that.basePersistDirectory != null) {
       return false;
     }
+
     return indexSpec != null ? indexSpec.equals(that.indexSpec) : that.indexSpec == null;
 
   }

extensions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java

@@ -80,7 +80,6 @@ public KafkaSupervisorSpec(
                             null,
                             null,
                             null,
-                            null,
                             null
                         );
     this.ioConfig = Preconditions.checkNotNull(ioConfig, "ioConfig");

extensions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorTuningConfig.java

@@ -25,8 +25,6 @@
 import org.joda.time.Duration;
 import org.joda.time.Period;
 
-import java.io.File;
-
 public class KafkaSupervisorTuningConfig extends KafkaTuningConfig
 {
   private final Integer workerThreads;
@@ -39,7 +37,6 @@ public KafkaSupervisorTuningConfig(
       @JsonProperty("maxRowsInMemory") Integer maxRowsInMemory,
       @JsonProperty("maxRowsPerSegment") Integer maxRowsPerSegment,
       @JsonProperty("intermediatePersistPeriod") Period intermediatePersistPeriod,
-      @JsonProperty("basePersistDirectory") File basePersistDirectory,
       @JsonProperty("maxPendingPersists") Integer maxPendingPersists,
       @JsonProperty("indexSpec") IndexSpec indexSpec,
       @JsonProperty("buildV9Directly") Boolean buildV9Directly,
@@ -57,7 +54,6 @@ public KafkaSupervisorTuningConfig(
         maxRowsInMemory,
         maxRowsPerSegment,
         intermediatePersistPeriod,
-        basePersistDirectory,
         maxPendingPersists,
         indexSpec,
         buildV9Directly,

extensions-core/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/KafkaIndexTaskTest.java

@@ -1355,7 +1355,6 @@ private KafkaIndexTask createTask(
         new Period("P1Y"),
         null,
         null,
-        null,
         buildV9Directly,
         reportParseExceptions,
         handoffConditionTimeout,

extensions-core/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/KafkaTuningConfigTest.java

@@ -28,8 +28,6 @@
 import org.junit.Assert;
 import org.junit.Test;
 
-import java.io.File;
-
 public class KafkaTuningConfigTest
 {
   private final ObjectMapper mapper;
@@ -55,7 +53,7 @@ public void testSerdeWithDefaults() throws Exception
         TuningConfig.class
     );
 
-    Assert.assertNotNull(config.getBasePersistDirectory());
+    Assert.assertNull(config.getBasePersistDirectory());
     Assert.assertEquals(75000, config.getMaxRowsInMemory());
     Assert.assertEquals(5_000_000, config.getMaxRowsPerSegment());
     Assert.assertEquals(new Period("PT10M"), config.getIntermediatePersistPeriod());
@@ -71,7 +69,6 @@ public void testSerdeWithNonDefaults() throws Exception
   {
     String jsonStr = "{\n"
                      + "  \"type\": \"kafka\",\n"
-                     + "  \"basePersistDirectory\": \"/tmp/xxx\",\n"
                      + "  \"maxRowsInMemory\": 100,\n"
                      + "  \"maxRowsPerSegment\": 100,\n"
                      + "  \"intermediatePersistPeriod\": \"PT1H\",\n"
@@ -91,7 +88,7 @@ public void testSerdeWithNonDefaults() throws Exception
         TuningConfig.class
     );
 
-    Assert.assertEquals(new File("/tmp/xxx"), config.getBasePersistDirectory());
+    Assert.assertNull(config.getBasePersistDirectory());
     Assert.assertEquals(100, config.getMaxRowsInMemory());
     Assert.assertEquals(100, config.getMaxRowsPerSegment());
     Assert.assertEquals(new Period("PT1H"), config.getIntermediatePersistPeriod());
@@ -104,13 +101,12 @@ public void testSerdeWithNonDefaults() throws Exception
   @Test
   public void testCopyOf() throws Exception
   {
-    KafkaTuningConfig original = new KafkaTuningConfig(1, 2, new Period("PT3S"), new File("/tmp/xxx"), 4, new IndexSpec(), true, true, 5L, null);
+    KafkaTuningConfig original = new KafkaTuningConfig(1, 2, new Period("PT3S"), 4, new IndexSpec(), true, true, 5L, null);
     KafkaTuningConfig copy = KafkaTuningConfig.copyOf(original);
 
     Assert.assertEquals(1, copy.getMaxRowsInMemory());
     Assert.assertEquals(2, copy.getMaxRowsPerSegment());
     Assert.assertEquals(new Period("PT3S"), copy.getIntermediatePersistPeriod());
-    Assert.assertEquals(new File("/tmp/xxx"), copy.getBasePersistDirectory());
     Assert.assertEquals(4, copy.getMaxPendingPersists());
     Assert.assertEquals(new IndexSpec(), copy.getIndexSpec());
     Assert.assertEquals(true, copy.getBuildV9Directly());

extensions-core/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java

@@ -85,7 +85,6 @@
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
-import java.io.File;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -169,7 +168,6 @@ public void setUp() throws Exception
         1000,
         50000,
         new Period("P1Y"),
-        new File("/test"),
         null,
         null,
         true,

extensions-core/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/supervisor/KafkaSupervisorTuningConfigTest.java

@@ -30,8 +30,6 @@
 import org.junit.Assert;
 import org.junit.Test;
 
-import java.io.File;
-
 public class KafkaSupervisorTuningConfigTest
 {
   private final ObjectMapper mapper;
@@ -57,7 +55,7 @@ public void testSerdeWithDefaults() throws Exception
         TuningConfig.class
     );
 
-    Assert.assertNotNull(config.getBasePersistDirectory());
+    Assert.assertNull(config.getBasePersistDirectory());
     Assert.assertEquals(75000, config.getMaxRowsInMemory());
     Assert.assertEquals(5_000_000, config.getMaxRowsPerSegment());
     Assert.assertEquals(new Period("PT10M"), config.getIntermediatePersistPeriod());
@@ -78,7 +76,6 @@ public void testSerdeWithNonDefaults() throws Exception
   {
     String jsonStr = "{\n"
                      + "  \"type\": \"kafka\",\n"
-                     + "  \"basePersistDirectory\": \"/tmp/xxx\",\n"
                      + "  \"maxRowsInMemory\": 100,\n"
                      + "  \"maxRowsPerSegment\": 100,\n"
                      + "  \"intermediatePersistPeriod\": \"PT1H\",\n"
@@ -103,7 +100,7 @@ public void testSerdeWithNonDefaults() throws Exception
         TuningConfig.class
     );
 
-    Assert.assertEquals(new File("/tmp/xxx"), config.getBasePersistDirectory());
+    Assert.assertNull(config.getBasePersistDirectory());
     Assert.assertEquals(100, config.getMaxRowsInMemory());
     Assert.assertEquals(100, config.getMaxRowsPerSegment());
     Assert.assertEquals(new Period("PT1H"), config.getIntermediatePersistPeriod());

indexing-service/src/main/java/io/druid/indexing/common/task/RealtimeIndexTask.java

@@ -283,8 +283,9 @@ public String getVersion(final Interval interval)
     DataSchema dataSchema = spec.getDataSchema();
     RealtimeIOConfig realtimeIOConfig = spec.getIOConfig();
     RealtimeTuningConfig tuningConfig = spec.getTuningConfig()
-                                            .withBasePersistDirectory(new File(toolbox.getTaskWorkDir(), "persist"))
-                                            .withVersioningPolicy(versioningPolicy);
+                                            .withBasePersistDirectorAndVersioningPolicy(
+                                                new File(toolbox.getTaskWorkDir(), "persist"),
+                                                versioningPolicy);
 
     final FireDepartment fireDepartment = new FireDepartment(
         dataSchema,

indexing-service/src/test/java/io/druid/indexing/common/task/RealtimeIndexTaskTest.java

@@ -896,23 +896,16 @@ private RealtimeIndexTask makeRealtimeTask(final String taskId, boolean reportPa
         null,
         null
     );
-    RealtimeTuningConfig realtimeTuningConfig = new RealtimeTuningConfig(
-        1000,
-        new Period("P1Y"),
-        new Period("PT10M"),
-        null,
-        null,
-        new ServerTimeRejectionPolicyFactory(),
-        null,
-        null,
-        null,
-        buildV9Directly,
-        0,
-        0,
-        reportParseExceptions,
-        handoffTimeout,
-        null
-    );
+    RealtimeTuningConfig realtimeTuningConfig = new RealtimeTuningConfig.Builder()
+        .withMaxRowsInMemory(1000)
+        .withIntermediatePersistPeriod(new Period("P1Y"))
+        .withWindowPeriod(new Period("PT10M"))
+        .withRejectionPolicyFactory(new ServerTimeRejectionPolicyFactory())
+        .withBuildV9Directly(buildV9Directly)
+        .withReportParseExceptions(reportParseExceptions)
+        .withHandoffConditionTimeout(handoffTimeout)
+        .build();
+
     return new RealtimeIndexTask(
         taskId,
         null,

indexing-service/src/test/java/io/druid/indexing/common/task/TaskSerdeTest.java

@@ -489,23 +489,14 @@ public Plumber findPlumber(
                 null
             ),
 
-            new RealtimeTuningConfig(
-                1,
-                new Period("PT10M"),
-                null,
-                null,
-                null,
-                null,
-                1,
-                NoneShardSpec.instance(),
-                indexSpec,
-                null,
-                0,
-                0,
-                true,
-                null,
-                null
-            )
+            new RealtimeTuningConfig.Builder()
+                .withMaxRowsInMemory(1)
+                .withIntermediatePersistPeriod(new Period("PT10M"))
+                .withMaxPendingPersists(1)
+                .withShardSpec(NoneShardSpec.instance())
+                .withIndexSpec(indexSpec)
+                .withReportParseExceptions(true)
+                .build()
         ),
         null
     );

indexing-service/src/test/java/io/druid/indexing/overlord/TaskLifecycleTest.java

@@ -1177,23 +1177,10 @@ private RealtimeIndexTask newRealtimeIndexTask()
         // PlumberSchool - Realtime Index Task always uses RealtimePlumber which is hardcoded in RealtimeIndexTask class
         null
     );
-    RealtimeTuningConfig realtimeTuningConfig = new RealtimeTuningConfig(
-        1000,
-        new Period("P1Y"),
-        null, //default window period of 10 minutes
-        null, // base persist dir ignored by Realtime Index task
-        null,
-        null,
-        null,
-        null,
-        null,
-        null,
-        0,
-        0,
-        null,
-        null,
-        null
-    );
+    RealtimeTuningConfig realtimeTuningConfig = new RealtimeTuningConfig.Builder()
+        .withMaxRowsInMemory(1000)
+        .withIntermediatePersistPeriod(new Period("P1Y"))
+        .build();
     FireDepartment fireDepartment = new FireDepartment(dataSchema, realtimeIOConfig, realtimeTuningConfig);
     return new RealtimeIndexTask(
         taskId,