From b576980aeb9797957fa950386a3ebe30b7e22dea Mon Sep 17 00:00:00 2001
From: Himanshu Gupta <g.himanshu@gmail.com>
Date: Thu, 23 May 2019 11:43:00 -0700
Subject: [PATCH] use memoized supplier for lazy singleton in
 SeekableStreamIndexTask.java

---
 .../seekablestream/SeekableStreamIndexTask.java  | 16 +++++-----------
 1 file changed, 5 insertions(+), 11 deletions(-)

diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTask.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTask.java
index 37d472bb59f8..11ed7d4d7649 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTask.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTask.java
@@ -24,6 +24,8 @@
 import com.google.common.base.Joiner;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
 import org.apache.druid.data.input.InputRow;
 import org.apache.druid.indexer.TaskStatus;
 import org.apache.druid.indexing.appenderator.ActionBasedSegmentAllocator;
@@ -77,8 +79,7 @@ public abstract class SeekableStreamIndexTask<PartitionIdType, SequenceOffsetTyp
   // Lazily initialized, to avoid calling it on the overlord when tasks are instantiated.
   // See https://github.com/apache/incubator-druid/issues/7724 for issues that can cause.
   // By the way, lazily init is synchronized because the runner may be needed in multiple threads.
-  private final Object runnerInitLock = new Object();
-  private volatile SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOffsetType> runner;
+  private final Supplier<SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOffsetType>> runnerSupplier;
 
   public SeekableStreamIndexTask(
       final String id,
@@ -112,6 +113,7 @@ public SeekableStreamIndexTask(
     this.context = context;
     this.authorizerMapper = authorizerMapper;
     this.rowIngestionMetersFactory = rowIngestionMetersFactory;
+    this.runnerSupplier = Suppliers.memoize(this::createTaskRunner);
   }
 
   private static String makeTaskId(String dataSource, String type)
@@ -288,14 +290,6 @@ public Appenderator getAppenderator()
   @VisibleForTesting
   public SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOffsetType> getRunner()
   {
-    if (runner == null) {
-      synchronized (runnerInitLock) {
-        if (runner == null) {
-          runner = createTaskRunner();
-        }
-      }
-    }
-
-    return runner;
+    return runnerSupplier.get();
   }
 }