support both rejectionPolicy and rejectionPolicyFactory in serde

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

@@ -115,7 +115,8 @@ public class RealtimeIndexTask extends AbstractTask
       @JsonProperty("windowPeriod") Period windowPeriod,
       @JsonProperty("maxPendingPersists") int maxPendingPersists,
       @JsonProperty("segmentGranularity") Granularity segmentGranularity,
-      @JsonProperty("rejectionPolicy") RejectionPolicyFactory rejectionPolicyFactory
+      @JsonProperty("rejectionPolicy") RejectionPolicyFactory rejectionPolicy,
+      @JsonProperty("rejectionPolicyFactory") RejectionPolicyFactory rejectionPolicyFactory
   )
   {
     super(
@@ -142,7 +143,7 @@ public class RealtimeIndexTask extends AbstractTask
               windowPeriod,
               null,
               null,
-              rejectionPolicyFactory,
+              rejectionPolicy == null ? rejectionPolicyFactory : rejectionPolicy,
               maxPendingPersists,
               spec.getShardSpec()
           ),
@@ -315,6 +316,7 @@ public class RealtimeIndexTask extends AbstractTask
         null,
         null,
         null,
+        null,
         0
     );
 

indexing-service/src/test/java/io/druid/indexing/common/TestRealtimeTask.java

@@ -54,6 +54,7 @@ public class TestRealtimeTask extends RealtimeIndexTask
         null,
         1,
         null,
+        null,
         null
     );
     this.status = status;

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

@@ -207,6 +207,7 @@ public class TaskSerdeTest
         new Period("PT10M"),
         1,
         Granularity.HOUR,
+        null,
         null
     );
 

indexing-service/src/test/java/io/druid/indexing/worker/TaskAnnouncementTest.java

@@ -49,6 +49,7 @@ public class TaskAnnouncementTest
         new Period("PT10M"),
         1,
         Granularity.HOUR,
+        null,
         null
     );
     final TaskStatus status = TaskStatus.running(task.getId());

server/src/main/java/io/druid/segment/realtime/plumber/FlushingPlumberSchool.java

@@ -65,7 +65,8 @@ public class FlushingPlumberSchool extends RealtimePlumberSchool
       @JsonProperty("basePersistDirectory") File basePersistDirectory,
       @JsonProperty("segmentGranularity") Granularity segmentGranularity,
       @JsonProperty("versioningPolicy") VersioningPolicy versioningPolicy,
-      @JsonProperty("rejectionPolicy") RejectionPolicyFactory rejectionPolicyFactory,
+      @JsonProperty("rejectionPolicy") RejectionPolicyFactory rejectionPolicy,
+      @JsonProperty("rejectionPolicyFactory") RejectionPolicyFactory rejectionPolicyFactory,
       @JsonProperty("maxPendingPersists") int maxPendingPersists
   )
   {
@@ -81,6 +82,7 @@ public class FlushingPlumberSchool extends RealtimePlumberSchool
         basePersistDirectory,
         segmentGranularity,
         versioningPolicy,
+        rejectionPolicy,
         rejectionPolicyFactory,
         maxPendingPersists
     );

server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumberSchool.java

@@ -75,7 +75,8 @@ public class RealtimePlumberSchool implements PlumberSchool
       @JsonProperty("basePersistDirectory") File basePersistDirectory,
       @JsonProperty("segmentGranularity") Granularity segmentGranularity,
       @JsonProperty("versioningPolicy") VersioningPolicy versioningPolicy,
-      @JsonProperty("rejectionPolicy") RejectionPolicyFactory rejectionPolicyFactory,
+      @JsonProperty("rejectionPolicy") RejectionPolicyFactory rejectionPolicy,
+      @JsonProperty("rejectionPolicyFactory") RejectionPolicyFactory rejectionPolicyFactory,
       @JsonProperty("maxPendingPersists") int maxPendingPersists
   )
   {
@@ -90,7 +91,7 @@ public class RealtimePlumberSchool implements PlumberSchool
     this.basePersistDirectory = basePersistDirectory;
     this.segmentGranularity = segmentGranularity;
     this.versioningPolicy = versioningPolicy;
-    this.rejectionPolicyFactory = rejectionPolicyFactory;
+    this.rejectionPolicyFactory = (rejectionPolicy == null) ? rejectionPolicyFactory : rejectionPolicy;
     this.maxPendingPersists = maxPendingPersists;
   }
 

server/src/test/java/io/druid/segment/realtime/FireDepartmentTest.java

@@ -73,7 +73,7 @@ public class FireDepartmentTest
         new RealtimeIOConfig(
             null,
             new RealtimePlumberSchool(
-                null, null, null, null, null, null, null, null, null, null, null, null, 0
+                null, null, null, null, null, null, null, null, null, null, null, null, null, 0
             )
         ),
         new RealtimeTuningConfig(

server/src/test/java/io/druid/segment/realtime/plumber/RealtimePlumberSchoolTest.java

@@ -149,6 +149,7 @@ public class RealtimePlumberSchoolTest
         Granularity.HOUR,
         new IntervalStartVersioningPolicy(),
         new NoopRejectionPolicyFactory(),
+        null,
         0
     );