This is an automated email from the ASF dual-hosted git repository.
fjy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-druid.git
The following commit(s) were added to refs/heads/master by this push:
new 53b6467 SeekableStreamIndexTaskRunner: Lazy init of runner. (#7729)
53b6467 is described below
commit 53b6467fc83cd4a78d87b5fd1557c84b2a5b2513
Author: Gian Merlino <g...@imply.io>
AuthorDate: Wed May 22 21:13:57 2019 -0700
SeekableStreamIndexTaskRunner: Lazy init of runner. (#7729)
The main motivation is that this fixes #7724, by making it so the overlord
doesn't try to create a task runner and parser when all it really wants to
do is create a task object and serialize it.
---
.../druid/indexing/kafka/KafkaIndexTask.java | 6 +-
.../druid/indexing/kafka/KafkaIndexTaskTest.java | 67 ++++++++++++++++------
.../druid/indexing/kinesis/KinesisIndexTask.java | 3 +-
.../seekablestream/SeekableStreamIndexTask.java | 32 ++++++-----
4 files changed, 76 insertions(+), 32 deletions(-)
diff --git
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java
index df56cce..314f99c 100644
---
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java
+++
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java
@@ -131,18 +131,20 @@ public class KafkaIndexTask extends
SeekableStreamIndexTask<Integer, Long>
{
if (context != null &&
context.get(SeekableStreamSupervisor.IS_INCREMENTAL_HANDOFF_SUPPORTED) != null
&& ((boolean)
context.get(SeekableStreamSupervisor.IS_INCREMENTAL_HANDOFF_SUPPORTED))) {
+ //noinspection unchecked
return new IncrementalPublishingKafkaIndexTaskRunner(
this,
- parser,
+ dataSchema.getParser(),
authorizerMapper,
chatHandlerProvider,
savedParseExceptions,
rowIngestionMetersFactory
);
} else {
+ //noinspection unchecked
return new LegacyKafkaIndexTaskRunner(
this,
- parser,
+ dataSchema.getParser(),
authorizerMapper,
chatHandlerProvider,
savedParseExceptions,
diff --git
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
index 6b9a203..4c9b39b 100644
---
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
+++
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
@@ -195,7 +195,7 @@ import static org.apache.druid.query.QueryPlus.wrap;
public class KafkaIndexTaskTest
{
private static final Logger log = new Logger(KafkaIndexTaskTest.class);
- private static final ObjectMapper objectMapper = TestHelper.makeJsonMapper();
+ private static final ObjectMapper OBJECT_MAPPER = new
TestUtils().getTestObjectMapper();
private static final long POLL_RETRY_MS = 100;
private static TestingCluster zkServer;
@@ -204,6 +204,10 @@ public class KafkaIndexTaskTest
private static ListeningExecutorService taskExec;
private static int topicPostfix;
+ static {
+ new
KafkaIndexTaskModule().getJacksonModules().forEach(OBJECT_MAPPER::registerModule);
+ }
+
private final List<Task> runningTasks = new ArrayList<>();
private long handoffConditionTimeout = 0;
@@ -244,7 +248,7 @@ public class KafkaIndexTaskTest
private static final DataSchema DATA_SCHEMA = new DataSchema(
"test_ds",
- objectMapper.convertValue(
+ OBJECT_MAPPER.convertValue(
new StringInputRowParser(
new JSONParseSpec(
new TimestampSpec("timestamp", "iso", null),
@@ -272,7 +276,7 @@ public class KafkaIndexTaskTest
},
new UniformGranularitySpec(Granularities.DAY, Granularities.NONE, null),
null,
- objectMapper
+ OBJECT_MAPPER
);
private static List<ProducerRecord<byte[], byte[]>> generateRecords(String
topic)
@@ -730,10 +734,11 @@ public class KafkaIndexTaskTest
SegmentDescriptor desc7 = sd(task, "2013/P1D", 0);
Assert.assertEquals(ImmutableSet.of(desc1, desc2, desc3, desc4, desc5,
desc6, desc7), publishedDescriptors());
Assert.assertEquals(
- new KafkaDataSourceMetadata(
- new SeekableStreamEndSequenceNumbers<>(topic, ImmutableMap.of(0,
10L, 1, 2L))
- ),
-
metadataStorageCoordinator.getDataSourceMetadata(DATA_SCHEMA.getDataSource()));
+ new KafkaDataSourceMetadata(
+ new SeekableStreamEndSequenceNumbers<>(topic, ImmutableMap.of(0,
10L, 1, 2L))
+ ),
+
metadataStorageCoordinator.getDataSourceMetadata(DATA_SCHEMA.getDataSource())
+ );
Assert.assertEquals(ImmutableSet.of(desc1, desc2, desc3, desc4, desc5,
desc6, desc7), publishedDescriptors());
Assert.assertEquals(
@@ -2011,7 +2016,7 @@ public class KafkaIndexTaskTest
Assert.assertEquals(2, countEvents(task));
Assert.assertEquals(Status.READING, task.getRunner().getStatus());
- Map<Integer, Long> currentOffsets = objectMapper.readValue(
+ Map<Integer, Long> currentOffsets = OBJECT_MAPPER.readValue(
task.getRunner().pause().getEntity().toString(),
new TypeReference<Map<Integer, Long>>()
{
@@ -2147,7 +2152,7 @@ public class KafkaIndexTaskTest
final Map<String, Object> context = new HashMap<>();
context.put(
SeekableStreamSupervisor.CHECKPOINTS_CTX_KEY,
-
objectMapper.writerFor(KafkaSupervisor.CHECKPOINTS_TYPE_REF).writeValueAsString(sequences)
+
OBJECT_MAPPER.writerFor(KafkaSupervisor.CHECKPOINTS_TYPE_REF).writeValueAsString(sequences)
);
final KafkaIndexTask task = createTask(
@@ -2267,7 +2272,7 @@ public class KafkaIndexTaskTest
Assert.assertEquals(Status.READING, task.getRunner().getStatus());
//verify the 2 indexed records
- final QuerySegmentSpec firstInterval = objectMapper.readValue(
+ final QuerySegmentSpec firstInterval = OBJECT_MAPPER.readValue(
"\"2008/2010\"", QuerySegmentSpec.class
);
Iterable<ScanResultValue> scanResultValues = scanData(task, firstInterval);
@@ -2287,7 +2292,7 @@ public class KafkaIndexTaskTest
Assert.assertEquals(2, countEvents(task));
Assert.assertEquals(Status.READING, task.getRunner().getStatus());
- final QuerySegmentSpec rollbackedInterval = objectMapper.readValue(
+ final QuerySegmentSpec rollbackedInterval = OBJECT_MAPPER.readValue(
"\"2010/2012\"", QuerySegmentSpec.class
);
scanResultValues = scanData(task, rollbackedInterval);
@@ -2304,7 +2309,7 @@ public class KafkaIndexTaskTest
kafkaProducer.commitTransaction();
}
- final QuerySegmentSpec endInterval = objectMapper.readValue(
+ final QuerySegmentSpec endInterval = OBJECT_MAPPER.readValue(
"\"2008/2049\"", QuerySegmentSpec.class
);
Iterable<ScanResultValue> scanResultValues1 = scanData(task, endInterval);
@@ -2388,6 +2393,36 @@ public class KafkaIndexTaskTest
Assert.assertEquals(TaskState.SUCCESS, future.get().getStatusCode());
}
+ @Test
+ public void testSerde() throws Exception
+ {
+ // This is both a serde test and a regression test for
https://github.com/apache/incubator-druid/issues/7724.
+
+ final KafkaIndexTask task = createTask(
+ "taskid",
+ DATA_SCHEMA.withTransformSpec(
+ new TransformSpec(
+ null,
+ ImmutableList.of(new ExpressionTransform("beep", "nofunc()",
ExprMacroTable.nil()))
+ )
+ ),
+ new KafkaIndexTaskIOConfig(
+ 0,
+ "sequence",
+ new SeekableStreamStartSequenceNumbers<>(topic, ImmutableMap.of(),
ImmutableSet.of()),
+ new SeekableStreamEndSequenceNumbers<>(topic, ImmutableMap.of()),
+ ImmutableMap.of(),
+ KafkaSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS,
+ true,
+ null,
+ null
+ )
+ );
+
+ final Task task1 =
OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsBytes(task), Task.class);
+ Assert.assertEquals(task, task1);
+ }
+
private List<ScanResultValue> scanData(final Task task, QuerySegmentSpec
spec)
{
ScanQuery query = new Druids.ScanQueryBuilder().dataSource(
@@ -2513,7 +2548,7 @@ public class KafkaIndexTaskTest
if (!context.containsKey(SeekableStreamSupervisor.CHECKPOINTS_CTX_KEY)) {
final TreeMap<Integer, Map<Integer, Long>> checkpoints = new
TreeMap<>();
checkpoints.put(0,
ioConfig.getStartSequenceNumbers().getPartitionSequenceNumberMap());
- final String checkpointsJson = objectMapper
+ final String checkpointsJson = OBJECT_MAPPER
.writerFor(KafkaSupervisor.CHECKPOINTS_TYPE_REF)
.writeValueAsString(checkpoints);
context.put(SeekableStreamSupervisor.CHECKPOINTS_CTX_KEY,
checkpointsJson);
@@ -2530,7 +2565,7 @@ public class KafkaIndexTaskTest
null,
null,
rowIngestionMetersFactory,
- objectMapper
+ OBJECT_MAPPER
);
task.setPollRetryMs(POLL_RETRY_MS);
return task;
@@ -2544,7 +2579,7 @@ public class KafkaIndexTaskTest
dataSchema.getAggregators(),
dataSchema.getGranularitySpec(),
dataSchema.getTransformSpec(),
- objectMapper
+ OBJECT_MAPPER
);
}
@@ -2861,7 +2896,7 @@ public class KafkaIndexTaskTest
private IngestionStatsAndErrorsTaskReportData getTaskReportData() throws
IOException
{
- Map<String, TaskReport> taskReports = objectMapper.readValue(
+ Map<String, TaskReport> taskReports = OBJECT_MAPPER.readValue(
reportsFile,
new TypeReference<Map<String, TaskReport>>()
{
diff --git
a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java
b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java
index ff1847a..f3dfe3b 100644
---
a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java
+++
b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java
@@ -72,9 +72,10 @@ public class KinesisIndexTask extends
SeekableStreamIndexTask<String, String>
@Override
protected SeekableStreamIndexTaskRunner<String, String> createTaskRunner()
{
+ //noinspection unchecked
return new KinesisIndexTaskRunner(
this,
- parser,
+ dataSchema.getParser(),
authorizerMapper,
chatHandlerProvider,
savedParseExceptions,
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 833cc17..37d472b 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
@@ -25,7 +25,6 @@ import com.google.common.base.Joiner;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import org.apache.druid.data.input.InputRow;
-import org.apache.druid.data.input.impl.InputRowParser;
import org.apache.druid.indexer.TaskStatus;
import org.apache.druid.indexing.appenderator.ActionBasedSegmentAllocator;
import org.apache.druid.indexing.appenderator.ActionBasedUsedSegmentChecker;
@@ -57,7 +56,6 @@ import org.apache.druid.server.security.AuthorizerMapper;
import org.apache.druid.utils.CircularBuffer;
import javax.annotation.Nullable;
-import java.nio.ByteBuffer;
import java.util.Map;
@@ -67,9 +65,7 @@ public abstract class
SeekableStreamIndexTask<PartitionIdType, SequenceOffsetTyp
public static final long LOCK_ACQUIRE_TIMEOUT_SECONDS = 15;
private static final EmittingLogger log = new
EmittingLogger(SeekableStreamIndexTask.class);
- private final SeekableStreamIndexTaskRunner<PartitionIdType,
SequenceOffsetType> runner;
protected final DataSchema dataSchema;
- protected final InputRowParser<ByteBuffer> parser;
protected final SeekableStreamIndexTaskTuningConfig tuningConfig;
protected final SeekableStreamIndexTaskIOConfig<PartitionIdType,
SequenceOffsetType> ioConfig;
protected final Optional<ChatHandlerProvider> chatHandlerProvider;
@@ -78,6 +74,12 @@ public abstract class
SeekableStreamIndexTask<PartitionIdType, SequenceOffsetTyp
protected final RowIngestionMetersFactory rowIngestionMetersFactory;
protected final CircularBuffer<Throwable> savedParseExceptions;
+ // 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;
+
public SeekableStreamIndexTask(
final String id,
@Nullable final TaskResource taskResource,
@@ -99,7 +101,6 @@ public abstract class
SeekableStreamIndexTask<PartitionIdType, SequenceOffsetTyp
context
);
this.dataSchema = Preconditions.checkNotNull(dataSchema, "dataSchema");
- this.parser = Preconditions.checkNotNull((InputRowParser<ByteBuffer>)
dataSchema.getParser(), "parser");
this.tuningConfig = Preconditions.checkNotNull(tuningConfig,
"tuningConfig");
this.ioConfig = Preconditions.checkNotNull(ioConfig, "ioConfig");
this.chatHandlerProvider = Optional.fromNullable(chatHandlerProvider);
@@ -111,7 +112,6 @@ public abstract class
SeekableStreamIndexTask<PartitionIdType, SequenceOffsetTyp
this.context = context;
this.authorizerMapper = authorizerMapper;
this.rowIngestionMetersFactory = rowIngestionMetersFactory;
- this.runner = createTaskRunner();
}
private static String makeTaskId(String dataSource, String type)
@@ -130,7 +130,6 @@ public abstract class
SeekableStreamIndexTask<PartitionIdType, SequenceOffsetTyp
return StringUtils.format("%s_%s", type, dataSource);
}
-
@Override
public int getPriority()
{
@@ -164,7 +163,7 @@ public abstract class
SeekableStreamIndexTask<PartitionIdType, SequenceOffsetTyp
@Override
public TaskStatus run(final TaskToolbox toolbox)
{
- return runner.run(toolbox);
+ return getRunner().run(toolbox);
}
@Override
@@ -177,19 +176,19 @@ public abstract class
SeekableStreamIndexTask<PartitionIdType, SequenceOffsetTyp
public void stopGracefully(TaskConfig taskConfig)
{
if (taskConfig.isRestoreTasksOnRestart()) {
- runner.stopGracefully();
+ getRunner().stopGracefully();
}
}
@Override
public <T> QueryRunner<T> getQueryRunner(Query<T> query)
{
- if (runner.getAppenderator() == null) {
+ if (getRunner().getAppenderator() == null) {
// Not yet initialized, no data yet, just return a noop runner.
return new NoopQueryRunner<>();
}
- return (queryPlus, responseContext) ->
queryPlus.run(runner.getAppenderator(), responseContext);
+ return (queryPlus, responseContext) ->
queryPlus.run(getRunner().getAppenderator(), responseContext);
}
public Appenderator newAppenderator(FireDepartmentMetrics metrics,
TaskToolbox toolbox)
@@ -283,13 +282,20 @@ public abstract class
SeekableStreamIndexTask<PartitionIdType, SequenceOffsetTyp
@VisibleForTesting
public Appenderator getAppenderator()
{
- return runner.getAppenderator();
+ return getRunner().getAppenderator();
}
@VisibleForTesting
public SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOffsetType>
getRunner()
{
+ if (runner == null) {
+ synchronized (runnerInitLock) {
+ if (runner == null) {
+ runner = createTaskRunner();
+ }
+ }
+ }
+
return runner;
}
-
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org
For additional commands, e-mail: commits-h...@druid.apache.org
