[Backport] maxTotalRows should be checked in DataSourceCompactionConfig before setting targetCompactionSizeBytes (#7379)

jon-wei · web-flow · commit 1e503190037a · 2019-03-28T22:55:32.000-07:00
* maxTotalRows should be checked in DataSourceCompactionConfig before setting targetCompactionSizeBytes (#7368) * maxTotalRows should be checked in DataSourceCompactionConfig before setting targetCompactionSizeBytes * remove unnecessary default values * remove flacky test * fix build * Add comments * checkstyle fix * Fix
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/hadoop/ITHadoopIndexTest.java b/integration-tests/src/test/java/org/apache/druid/tests/hadoop/ITHadoopIndexTest.java
@@ -65,7 +65,7 @@ private void loadData(String hadoopDir)
 
     try {
       LOG.info("indexerFile name: [%s]", BATCH_TASK);
-      indexerSpec = getTaskAsString(BATCH_TASK);
+      indexerSpec = getResourceAsString(BATCH_TASK);
       indexerSpec = StringUtils.replace(indexerSpec, "%%HADOOP_TEST_PATH%%", hadoopDir);
     }
     catch (Exception e) {
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITBatchIndexTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITBatchIndexTest.java
@@ -56,7 +56,7 @@ void doIndexTestTest(
   {
     final String fullDatasourceName = dataSource + config.getExtraDatasourceNameSuffix();
     final String taskSpec = StringUtils.replace(
-        getTaskAsString(indexTaskFilePath),
+        getResourceAsString(indexTaskFilePath),
         "%%DATASOURCE%%",
         fullDatasourceName
     );
@@ -98,7 +98,7 @@ void doReindexTest(
     final String fullReindexDatasourceName = reindexDataSource + config.getExtraDatasourceNameSuffix();
 
     String taskSpec = StringUtils.replace(
-        getTaskAsString(reindexTaskFilePath),
+        getResourceAsString(reindexTaskFilePath),
         "%%DATASOURCE%%",
         fullBaseDatasourceName
     );
@@ -148,7 +148,7 @@ void doIndexTestSqlTest(
   {
     final String fullDatasourceName = dataSource + config.getExtraDatasourceNameSuffix();
     final String taskSpec = StringUtils.replace(
-        getTaskAsString(indexTaskFilePath),
+        getResourceAsString(indexTaskFilePath),
         "%%DATASOURCE%%",
         fullDatasourceName
     );
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITRealtimeIndexTaskTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITRealtimeIndexTaskTest.java
@@ -89,7 +89,7 @@ void doTest()
     try (final Closeable closeable = unloader(fullDatasourceName)) {
       // the task will run for 3 minutes and then shutdown itself
       String task = setShutOffTime(
-          getTaskAsString(getTaskResource()),
+          getResourceAsString(getTaskResource()),
           DateTimes.utc(System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(3))
       );
       task = StringUtils.replace(task, "%%DATASOURCE%%", fullDatasourceName);
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractIndexerTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractIndexerTest.java
@@ -112,7 +112,7 @@ protected void waitForAllTasksToComplete()
     );
   }
 
-  protected String getTaskAsString(String file) throws IOException
+  protected String getResourceAsString(String file) throws IOException
   {
     final InputStream inputStream = ITRealtimeIndexTaskTest.class.getResourceAsStream(file);
     try {
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITCompactionTaskTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITCompactionTaskTest.java
@@ -131,7 +131,7 @@ public void testCompactionWithKeepSegmentGranularity() throws Exception
 
   private void loadData() throws Exception
   {
-    String taskSpec = getTaskAsString(INDEX_TASK);
+    String taskSpec = getResourceAsString(INDEX_TASK);
     taskSpec = StringUtils.replace(taskSpec, "%%DATASOURCE%%", fullDatasourceName);
     final String taskID = indexer.submitTask(taskSpec);
     LOG.info("TaskID for loading index task %s", taskID);
@@ -145,7 +145,7 @@ private void loadData() throws Exception
 
   private void compactData(boolean keepSegmentGranularity) throws Exception
   {
-    final String template = getTaskAsString(COMPACTION_TASK);
+    final String template = getResourceAsString(COMPACTION_TASK);
     String taskSpec =
         StringUtils.replace(template, "${KEEP_SEGMENT_GRANULARITY}", Boolean.toString(keepSegmentGranularity));
     taskSpec = StringUtils.replace(taskSpec, "%%DATASOURCE%%", fullDatasourceName);
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaIndexingServiceTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaIndexingServiceTest.java
@@ -152,7 +152,7 @@ public void testKafka()
       consumerProperties.put("bootstrap.servers", config.getKafkaInternalHost());
       addFilteredProperties(consumerProperties);
 
-      spec = getTaskAsString(INDEXER_FILE);
+      spec = getResourceAsString(INDEXER_FILE);
       spec = StringUtils.replace(spec, "%%DATASOURCE%%", fullDatasourceName);
       spec = StringUtils.replace(spec, "%%TOPIC%%", TOPIC_NAME);
       spec = StringUtils.replace(spec, "%%CONSUMER_PROPERTIES%%", jsonMapper.writeValueAsString(consumerProperties));
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaTest.java
@@ -144,7 +144,6 @@ public void testKafka()
     catch (Exception e) {
       throw new ISE(e, "could not create kafka topic");
     }
-
     // set up kafka producer
     Properties properties = new Properties();
     addFilteredProperties(properties);
@@ -211,7 +210,7 @@ public void testKafka()
 
       addFilteredProperties(consumerProperties);
 
-      indexerSpec = getTaskAsString(INDEXER_FILE);
+      indexerSpec = getResourceAsString(INDEXER_FILE);
       indexerSpec = StringUtils.replace(indexerSpec, "%%DATASOURCE%%", fullDatasourceName);
       indexerSpec = StringUtils.replace(indexerSpec, "%%TOPIC%%", TOPIC_NAME);
       indexerSpec = StringUtils.replace(indexerSpec, "%%COUNT%%", Integer.toString(num_events));
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITNestedQueryPushDownTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITNestedQueryPushDownTest.java
@@ -97,7 +97,7 @@ public void testIndexData()
 
   private void loadData() throws Exception
   {
-    String taskSpec = getTaskAsString(WIKITICKER_INDEX_TASK);
+    String taskSpec = getResourceAsString(WIKITICKER_INDEX_TASK);
     taskSpec = StringUtils.replace(taskSpec, "%%DATASOURCE%%", fullDatasourceName);
     final String taskID = indexer.submitTask(taskSpec);
     LOG.info("TaskID for loading index task %s", taskID);
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITUnionQueryTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITUnionQueryTest.java
@@ -93,7 +93,7 @@ public void testUnionQuery() throws IOException
     try {
       // Load 4 datasources with same dimensions
       String task = setShutOffTime(
-          getTaskAsString(UNION_TASK_RESOURCE),
+          getResourceAsString(UNION_TASK_RESOURCE),
           DateTimes.utc(System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(3))
       );
       List<String> taskIDs = new ArrayList<>();
diff --git a/server/src/main/java/org/apache/druid/client/indexing/ClientCompactQuery.java b/server/src/main/java/org/apache/druid/client/indexing/ClientCompactQuery.java
@@ -99,4 +99,17 @@ public Map<String, Object> getContext()
   {
     return context;
   }
+
+  @Override
+  public String toString()
+  {
+    return "ClientCompactQuery{" +
+           "dataSource='" + dataSource + '\'' +
+           ", segments=" + segments +
+           ", keepSegmentGranularity=" + keepSegmentGranularity +
+           ", targetCompactionSizeBytes=" + targetCompactionSizeBytes +
+           ", tuningConfig=" + tuningConfig +
+           ", context=" + context +
+           '}';
+  }
 }
diff --git a/server/src/main/java/org/apache/druid/client/indexing/ClientCompactQueryTuningConfig.java b/server/src/main/java/org/apache/druid/client/indexing/ClientCompactQueryTuningConfig.java
@@ -29,20 +29,18 @@
 
 public class ClientCompactQueryTuningConfig
 {
-  // These default values should be synchronized with those of IndexTuningConfig
-  private static final int DEFAULT_MAX_ROWS_IN_MEMORY = 75_000;
-  private static final int DEFAULT_MAX_TOTAL_ROWS = 20_000_000;
-  private static final IndexSpec DEFAULT_INDEX_SPEC = new IndexSpec();
-  private static final int DEFAULT_MAX_PENDING_PERSISTS = 0;
-  private static final long DEFAULT_PUSH_TIMEOUT = 0;
-
   @Nullable
   private final Integer maxRowsPerSegment;
-  private final int maxRowsInMemory;
-  private final int maxTotalRows;
+  @Nullable
+  private final Integer maxRowsInMemory;
+  @Nullable
+  private final Integer maxTotalRows;
+  @Nullable
   private final IndexSpec indexSpec;
-  private final int maxPendingPersists;
-  private final long pushTimeout;
+  @Nullable
+  private final Integer maxPendingPersists;
+  @Nullable
+  private final Long pushTimeout;
 
   public static ClientCompactQueryTuningConfig from(
       @Nullable UserCompactTuningConfig userCompactTuningConfig,
@@ -70,11 +68,11 @@ public ClientCompactQueryTuningConfig(
   )
   {
     this.maxRowsPerSegment = maxRowsPerSegment;
-    this.maxRowsInMemory = maxRowsInMemory == null ? DEFAULT_MAX_ROWS_IN_MEMORY : maxRowsInMemory;
-    this.maxTotalRows = maxTotalRows == null ? DEFAULT_MAX_TOTAL_ROWS : maxTotalRows;
-    this.indexSpec = indexSpec == null ? DEFAULT_INDEX_SPEC : indexSpec;
-    this.maxPendingPersists = maxPendingPersists == null ? DEFAULT_MAX_PENDING_PERSISTS : maxPendingPersists;
-    this.pushTimeout = pushTimeout == null ? DEFAULT_PUSH_TIMEOUT : pushTimeout;
+    this.maxRowsInMemory = maxRowsInMemory;
+    this.maxTotalRows = maxTotalRows;
+    this.indexSpec = indexSpec;
+    this.maxPendingPersists = maxPendingPersists;
+    this.pushTimeout = pushTimeout;
   }
 
   @JsonProperty
@@ -91,31 +89,36 @@ public Integer getMaxRowsPerSegment()
   }
 
   @JsonProperty
-  public int getMaxRowsInMemory()
+  @Nullable
+  public Integer getMaxRowsInMemory()
   {
     return maxRowsInMemory;
   }
 
   @JsonProperty
-  public int getMaxTotalRows()
+  @Nullable
+  public Integer getMaxTotalRows()
   {
     return maxTotalRows;
   }
 
   @JsonProperty
+  @Nullable
   public IndexSpec getIndexSpec()
   {
     return indexSpec;
   }
 
   @JsonProperty
-  public int getMaxPendingPersists()
+  @Nullable
+  public Integer getMaxPendingPersists()
   {
     return maxPendingPersists;
   }
 
   @JsonProperty
-  public long getPushTimeout()
+  @Nullable
+  public Long getPushTimeout()
   {
     return pushTimeout;
   }
@@ -130,25 +133,18 @@ public boolean equals(Object o)
       return false;
     }
     ClientCompactQueryTuningConfig that = (ClientCompactQueryTuningConfig) o;
-    return maxRowsInMemory == that.maxRowsInMemory &&
-           maxTotalRows == that.maxTotalRows &&
-           maxPendingPersists == that.maxPendingPersists &&
-           pushTimeout == that.pushTimeout &&
-           Objects.equals(maxRowsPerSegment, that.maxRowsPerSegment) &&
-           Objects.equals(indexSpec, that.indexSpec);
+    return Objects.equals(maxRowsPerSegment, that.maxRowsPerSegment) &&
+           Objects.equals(maxRowsInMemory, that.maxRowsInMemory) &&
+           Objects.equals(maxTotalRows, that.maxTotalRows) &&
+           Objects.equals(indexSpec, that.indexSpec) &&
+           Objects.equals(maxPendingPersists, that.maxPendingPersists) &&
+           Objects.equals(pushTimeout, that.pushTimeout);
   }
 
   @Override
   public int hashCode()
   {
-    return Objects.hash(
-        maxRowsPerSegment,
-        maxRowsInMemory,
-        maxTotalRows,
-        indexSpec,
-        maxPendingPersists,
-        pushTimeout
-    );
+    return Objects.hash(maxRowsPerSegment, maxRowsInMemory, maxTotalRows, indexSpec, maxPendingPersists, pushTimeout);
   }
 
   @Override
diff --git a/server/src/main/java/org/apache/druid/client/indexing/HttpIndexingServiceClient.java b/server/src/main/java/org/apache/druid/client/indexing/HttpIndexingServiceClient.java
@@ -22,6 +22,7 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import com.google.common.base.Throwables;
 import com.google.inject.Inject;
 import org.apache.druid.discovery.DruidLeaderClient;
@@ -93,7 +94,7 @@ public String compactSegments(
       boolean keepSegmentGranularity,
       @Nullable Long targetCompactionSizeBytes,
       int compactionTaskPriority,
-      @Nullable ClientCompactQueryTuningConfig tuningConfig,
+      ClientCompactQueryTuningConfig tuningConfig,
       @Nullable Map<String, Object> context
   )
   {
@@ -125,14 +126,20 @@ public String runTask(Object taskObject)
   {
     try {
       final FullResponseHolder response = druidLeaderClient.go(
-          druidLeaderClient.makeRequest(
-              HttpMethod.POST,
-              "/druid/indexer/v1/task"
-          ).setContent(MediaType.APPLICATION_JSON, jsonMapper.writeValueAsBytes(taskObject))
+          druidLeaderClient.makeRequest(HttpMethod.POST, "/druid/indexer/v1/task")
+                           .setContent(MediaType.APPLICATION_JSON, jsonMapper.writeValueAsBytes(taskObject))
       );
 
       if (!response.getStatus().equals(HttpResponseStatus.OK)) {
-        throw new ISE("Failed to post task[%s]", taskObject);
+        if (!Strings.isNullOrEmpty(response.getContent())) {
+          throw new ISE(
+              "Failed to post task[%s] with error[%s].",
+              taskObject,
+              response.getContent()
+          );
+        } else {
+          throw new ISE("Failed to post task[%s]. Please check overlord log", taskObject);
+        }
       }
 
       final Map<String, Object> resultMap = jsonMapper.readValue(
@@ -154,10 +161,7 @@ public String killTask(String taskId)
       final FullResponseHolder response = druidLeaderClient.go(
           druidLeaderClient.makeRequest(
               HttpMethod.POST,
-              StringUtils.format(
-                  "/druid/indexer/v1/task/%s/shutdown",
-                  StringUtils.urlEncode(taskId)
-              )
+              StringUtils.format("/druid/indexer/v1/task/%s/shutdown", StringUtils.urlEncode(taskId))
           )
       );
 
@@ -289,7 +293,10 @@ public TaskPayloadResponse getTaskPayload(String taskId)
   {
     try {
       final FullResponseHolder responseHolder = druidLeaderClient.go(
-          druidLeaderClient.makeRequest(HttpMethod.GET, StringUtils.format("/druid/indexer/v1/task/%s", taskId))
+          druidLeaderClient.makeRequest(
+              HttpMethod.GET,
+              StringUtils.format("/druid/indexer/v1/task/%s", StringUtils.urlEncode(taskId))
+          )
       );
 
       return jsonMapper.readValue(
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/helper/DruidCoordinatorSegmentCompactor.java b/server/src/main/java/org/apache/druid/server/coordinator/helper/DruidCoordinatorSegmentCompactor.java
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigTest.java b/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigTest.java

Original file line number	Diff line number	Diff line change
`@@ -65,7 +65,7 @@ private void loadData(String hadoopDir)`
`65`	`65`
`66`	`66`	`try {`
`67`	`67`	`LOG.info("indexerFile name: [%s]", BATCH_TASK);`
`68`		`- indexerSpec = getTaskAsString(BATCH_TASK);`
	`68`	`+ indexerSpec = getResourceAsString(BATCH_TASK);`
`69`	`69`	`indexerSpec = StringUtils.replace(indexerSpec, "%%HADOOP_TEST_PATH%%", hadoopDir);`
`70`	`70`	`}`
`71`	`71`	`catch (Exception e) {`
Original file line number	Diff line number	Diff line change
`@@ -112,7 +112,7 @@ protected void waitForAllTasksToComplete()`
`112`	`112`	`);`
`113`	`113`	`}`
`114`	`114`
`115`		`- protected String getTaskAsString(String file) throws IOException`
	`115`	`+ protected String getResourceAsString(String file) throws IOException`
`116`	`116`	`{`
`117`	`117`	`final InputStream inputStream = ITRealtimeIndexTaskTest.class.getResourceAsStream(file);`
`118`	`118`	`try {`
Original file line number	Diff line number	Diff line change
`@@ -97,7 +97,7 @@ public void testIndexData()`
`97`	`97`
`98`	`98`	`private void loadData() throws Exception`
`99`	`99`	`{`
`100`		`- String taskSpec = getTaskAsString(WIKITICKER_INDEX_TASK);`
	`100`	`+ String taskSpec = getResourceAsString(WIKITICKER_INDEX_TASK);`
`101`	`101`	`taskSpec = StringUtils.replace(taskSpec, "%%DATASOURCE%%", fullDatasourceName);`
`102`	`102`	`final String taskID = indexer.submitTask(taskSpec);`
`103`	`103`	`LOG.info("TaskID for loading index task %s", taskID);`
Original file line number	Diff line number	Diff line change
`@@ -99,4 +99,17 @@ public Map<String, Object> getContext()`
`99`	`99`	`{`
`100`	`100`	`return context;`
`101`	`101`	`}`
	`102`	`+`
	`103`	`+ @Override`
	`104`	`+ public String toString()`
	`105`	`+ {`
	`106`	`+ return "ClientCompactQuery{" +`
	`107`	`+ "dataSource='" + dataSource + '\'' +`
	`108`	`+ ", segments=" + segments +`
	`109`	`+ ", keepSegmentGranularity=" + keepSegmentGranularity +`
	`110`	`+ ", targetCompactionSizeBytes=" + targetCompactionSizeBytes +`
	`111`	`+ ", tuningConfig=" + tuningConfig +`
	`112`	`+ ", context=" + context +`
	`113`	`+ '}';`
	`114`	`+ }`
`102`	`115`	`}`