Allow reordered segment allocation in kafka indexing service (apache#…

…5805) * Allow reordered segment allocation in kafka indexing service * address comments * fix a bug
jihoonson · Jul 5, 2018 · 504ca80 · 504ca80
1 parent b07b3af
commit 504ca80
Show file tree

Hide file tree

Showing 7 changed files with 577 additions and 276 deletions.
diff --git a/indexing-service/src/test/java/io/druid/indexing/common/task/IndexTaskTest.java b/indexing-service/src/test/java/io/druid/indexing/common/task/IndexTaskTest.java
@@ -397,12 +397,12 @@ public void testAppendToExisting() throws Exception
 
     Assert.assertEquals("test", segments.get(0).getDataSource());
     Assert.assertEquals(Intervals.of("2014/P1D"), segments.get(0).getInterval());
-    Assert.assertTrue(segments.get(0).getShardSpec().getClass().equals(NumberedShardSpec.class));
+    Assert.assertEquals(NumberedShardSpec.class, segments.get(0).getShardSpec().getClass());
     Assert.assertEquals(0, segments.get(0).getShardSpec().getPartitionNum());
 
     Assert.assertEquals("test", segments.get(1).getDataSource());
     Assert.assertEquals(Intervals.of("2014/P1D"), segments.get(1).getInterval());
-    Assert.assertTrue(segments.get(1).getShardSpec().getClass().equals(NumberedShardSpec.class));
+    Assert.assertEquals(NumberedShardSpec.class, segments.get(1).getShardSpec().getClass());
     Assert.assertEquals(1, segments.get(1).getShardSpec().getPartitionNum());
   }
 

diff --git a/java-util/src/main/java/io/druid/java/util/common/Pair.java b/java-util/src/main/java/io/druid/java/util/common/Pair.java
@@ -21,25 +21,28 @@
 
 import com.google.common.base.Function;
 
+import javax.annotation.Nullable;
 import java.util.Comparator;
 
 /**
  */
 public class Pair<T1, T2>
 {
 
-  public static <T1, T2> Pair<T1, T2> of(T1 lhs, T2 rhs)
+  public static <T1, T2> Pair<T1, T2> of(@Nullable T1 lhs, @Nullable T2 rhs)
   {
     return new Pair<>(lhs, rhs);
   }
 
+  @Nullable
   public final T1 lhs;
 
+  @Nullable
   public final T2 rhs;
 
   public Pair(
-      T1 lhs,
-      T2 rhs
+      @Nullable T1 lhs,
+      @Nullable T2 rhs
   )
   {
     this.lhs = lhs;