jihoonson commented on a change in pull request #6431: Add Kinesis Indexing
Service to core Druid
URL: https://github.com/apache/incubator-druid/pull/6431#discussion_r241624398
##########
File path:
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java
##########
@@ -208,840 +72,63 @@ public IncrementalPublishingKafkaIndexTaskRunner(
RowIngestionMetersFactory rowIngestionMetersFactory
)
{
+ super(
+ task,
+ parser,
+ authorizerMapper,
+ chatHandlerProvider,
+ savedParseExceptions,
+ rowIngestionMetersFactory
+ );
this.task = task;
- this.ioConfig = task.getIOConfig();
this.tuningConfig = task.getTuningConfig();
- this.parser = parser;
- this.authorizerMapper = authorizerMapper;
- this.chatHandlerProvider = chatHandlerProvider;
- this.savedParseExceptions = savedParseExceptions;
- this.topic = ioConfig.getStartPartitions().getTopic();
- this.rowIngestionMeters =
rowIngestionMetersFactory.createRowIngestionMeters();
-
- this.endOffsets = new
ConcurrentHashMap<>(ioConfig.getEndPartitions().getPartitionOffsetMap());
- this.sequences = new CopyOnWriteArrayList<>();
- this.ingestionState = IngestionState.NOT_STARTED;
-
- resetNextCheckpointTime();
}
@Override
- public TaskStatus run(TaskToolbox toolbox)
- {
- try {
- return runInternal(toolbox);
- }
- catch (Exception e) {
- log.error(e, "Encountered exception while running task.");
- final String errorMsg = Throwables.getStackTraceAsString(e);
-
toolbox.getTaskReportFileWriter().write(getTaskCompletionReports(errorMsg));
- return TaskStatus.failure(
- task.getId(),
- errorMsg
- );
- }
- }
-
- private TaskStatus runInternal(TaskToolbox toolbox) throws Exception
- {
- log.info("Starting up!");
-
- startTime = DateTimes.nowUtc();
- status = Status.STARTING;
- this.toolbox = toolbox;
-
- if (!restoreSequences()) {
- final TreeMap<Integer, Map<Integer, Long>> checkpoints =
getCheckPointsFromContext(toolbox, task);
- if (checkpoints != null) {
- Iterator<Entry<Integer, Map<Integer, Long>>> sequenceOffsets =
checkpoints.entrySet().iterator();
- Map.Entry<Integer, Map<Integer, Long>> previous =
sequenceOffsets.next();
- while (sequenceOffsets.hasNext()) {
- Map.Entry<Integer, Map<Integer, Long>> current =
sequenceOffsets.next();
- sequences.add(new SequenceMetadata(
- previous.getKey(),
- StringUtils.format("%s_%s", ioConfig.getBaseSequenceName(),
previous.getKey()),
- previous.getValue(),
- current.getValue(),
- true
- ));
- previous = current;
- }
- sequences.add(new SequenceMetadata(
- previous.getKey(),
- StringUtils.format("%s_%s", ioConfig.getBaseSequenceName(),
previous.getKey()),
- previous.getValue(),
- endOffsets,
- false
- ));
- } else {
- sequences.add(new SequenceMetadata(
- 0,
- StringUtils.format("%s_%s", ioConfig.getBaseSequenceName(), 0),
- ioConfig.getStartPartitions().getPartitionOffsetMap(),
- endOffsets,
- false
- ));
- }
- }
- log.info("Starting with sequences: %s", sequences);
-
- if (chatHandlerProvider.isPresent()) {
- log.info("Found chat handler of class[%s]",
chatHandlerProvider.get().getClass().getName());
- chatHandlerProvider.get().register(task.getId(), this, false);
- } else {
- log.warn("No chat handler detected");
- }
-
- runThread = Thread.currentThread();
-
- // Set up FireDepartmentMetrics
- final FireDepartment fireDepartmentForMetrics = new FireDepartment(
- task.getDataSchema(),
- new RealtimeIOConfig(null, null, null),
- null
- );
- fireDepartmentMetrics = fireDepartmentForMetrics.getMetrics();
- toolbox.getMonitorScheduler()
- .addMonitor(TaskRealtimeMetricsMonitorBuilder.build(task,
fireDepartmentForMetrics, rowIngestionMeters));
-
- final String lookupTier =
task.getContextValue(RealtimeIndexTask.CTX_KEY_LOOKUP_TIER);
- LookupNodeService lookupNodeService = lookupTier == null ?
- toolbox.getLookupNodeService() :
- new LookupNodeService(lookupTier);
- DiscoveryDruidNode discoveryDruidNode = new DiscoveryDruidNode(
- toolbox.getDruidNode(),
- NodeType.PEON,
- ImmutableMap.of(
- toolbox.getDataNodeService().getName(),
toolbox.getDataNodeService(),
- lookupNodeService.getName(), lookupNodeService
- )
- );
-
- Throwable caughtExceptionOuter = null;
- try (final KafkaConsumer<byte[], byte[]> consumer = task.newConsumer()) {
- toolbox.getDataSegmentServerAnnouncer().announce();
- toolbox.getDruidNodeAnnouncer().announce(discoveryDruidNode);
-
- appenderator = task.newAppenderator(fireDepartmentMetrics, toolbox);
- driver = task.newDriver(appenderator, toolbox, fireDepartmentMetrics);
-
- final String topic = ioConfig.getStartPartitions().getTopic();
-
- // Start up, set up initial offsets.
- final Object restoredMetadata = driver.startJob();
- if (restoredMetadata == null) {
- // no persist has happened so far
- // so either this is a brand new task or replacement of a failed task
-
Preconditions.checkState(sequences.get(0).startOffsets.entrySet().stream().allMatch(
- partitionOffsetEntry -> Longs.compare(
- partitionOffsetEntry.getValue(),
- ioConfig.getStartPartitions()
- .getPartitionOffsetMap()
- .get(partitionOffsetEntry.getKey())
- ) >= 0
- ), "Sequence offsets are not compatible with start offsets of task");
- nextOffsets.putAll(sequences.get(0).startOffsets);
- } else {
- final Map<String, Object> restoredMetadataMap = (Map) restoredMetadata;
- final KafkaPartitions restoredNextPartitions =
toolbox.getObjectMapper().convertValue(
- restoredMetadataMap.get(METADATA_NEXT_PARTITIONS),
- KafkaPartitions.class
- );
- nextOffsets.putAll(restoredNextPartitions.getPartitionOffsetMap());
-
- // Sanity checks.
- if
(!restoredNextPartitions.getTopic().equals(ioConfig.getStartPartitions().getTopic()))
{
- throw new ISE(
- "WTF?! Restored topic[%s] but expected topic[%s]",
- restoredNextPartitions.getTopic(),
- ioConfig.getStartPartitions().getTopic()
- );
- }
-
- if
(!nextOffsets.keySet().equals(ioConfig.getStartPartitions().getPartitionOffsetMap().keySet()))
{
- throw new ISE(
- "WTF?! Restored partitions[%s] but expected partitions[%s]",
- nextOffsets.keySet(),
- ioConfig.getStartPartitions().getPartitionOffsetMap().keySet()
- );
- }
- // sequences size can be 0 only when all sequences got published and
task stopped before it could finish
- // which is super rare
- if (sequences.size() == 0 || sequences.get(sequences.size() -
1).isCheckpointed()) {
- this.endOffsets.putAll(sequences.size() == 0
- ? nextOffsets
- : sequences.get(sequences.size() -
1).getEndOffsets());
- log.info("End offsets changed to [%s]", endOffsets);
- }
- }
-
- // Set up committer.
- final Supplier<Committer> committerSupplier = () -> {
- final Map<Integer, Long> snapshot = ImmutableMap.copyOf(nextOffsets);
- lastPersistedOffsets.clear();
- lastPersistedOffsets.putAll(snapshot);
-
- return new Committer()
- {
- @Override
- public Object getMetadata()
- {
- return ImmutableMap.of(
- METADATA_NEXT_PARTITIONS, new KafkaPartitions(
- ioConfig.getStartPartitions().getTopic(),
- snapshot
- )
- );
- }
-
- @Override
- public void run()
- {
- // Do nothing.
- }
- };
- };
-
- // restart publishing of sequences (if any)
- maybePersistAndPublishSequences(committerSupplier);
-
- Set<Integer> assignment = assignPartitionsAndSeekToNext(consumer, topic);
-
- ingestionState = IngestionState.BUILD_SEGMENTS;
-
- // Main loop.
- // Could eventually support leader/follower mode (for keeping replicas
more in sync)
- boolean stillReading = !assignment.isEmpty();
- status = Status.READING;
- Throwable caughtExceptionInner = null;
- try {
- while (stillReading) {
- if (possiblyPause()) {
- // The partition assignments may have changed while paused by a
call to setEndOffsets() so reassign
- // partitions upon resuming. This is safe even if the end offsets
have not been modified.
- assignment = assignPartitionsAndSeekToNext(consumer, topic);
-
- if (assignment.isEmpty()) {
- log.info("All partitions have been fully read");
- publishOnStop.set(true);
- stopRequested.set(true);
- }
- }
-
- // if stop is requested or task's end offset is set by call to
setEndOffsets method with finish set to true
- if (stopRequested.get() || sequences.get(sequences.size() -
1).isCheckpointed()) {
- status = Status.PUBLISHING;
- }
-
- if (stopRequested.get()) {
- break;
- }
-
- if (backgroundThreadException != null) {
- throw new RuntimeException(backgroundThreadException);
- }
-
- checkPublishAndHandoffFailure();
-
- maybePersistAndPublishSequences(committerSupplier);
-
- // The retrying business is because the KafkaConsumer throws
OffsetOutOfRangeException if the seeked-to
- // offset is not present in the topic-partition. This can happen if
we're asking a task to read from data
- // that has not been written yet (which is totally legitimate). So
let's wait for it to show up.
- ConsumerRecords<byte[], byte[]> records = ConsumerRecords.empty();
- try {
- records = consumer.poll(KafkaIndexTask.POLL_TIMEOUT_MILLIS);
- }
- catch (OffsetOutOfRangeException e) {
- log.warn("OffsetOutOfRangeException with message [%s]",
e.getMessage());
- possiblyResetOffsetsOrWait(e.offsetOutOfRangePartitions(),
consumer, toolbox);
- stillReading = !assignment.isEmpty();
- }
-
- SequenceMetadata sequenceToCheckpoint = null;
- for (ConsumerRecord<byte[], byte[]> record : records) {
- log.trace(
- "Got topic[%s] partition[%d] offset[%,d].",
- record.topic(),
- record.partition(),
- record.offset()
- );
-
- if (record.offset() < endOffsets.get(record.partition())) {
- if (record.offset() != nextOffsets.get(record.partition())) {
- if (ioConfig.isSkipOffsetGaps()) {
- log.warn(
- "Skipped to offset[%,d] after offset[%,d] in
partition[%d].",
- record.offset(),
- nextOffsets.get(record.partition()),
- record.partition()
- );
- } else {
- throw new ISE(
- "WTF?! Got offset[%,d] after offset[%,d] in
partition[%d].",
- record.offset(),
- nextOffsets.get(record.partition()),
- record.partition()
- );
- }
- }
-
- try {
- final byte[] valueBytes = record.value();
- final List<InputRow> rows = valueBytes == null
- ? Utils.nullableListOf((InputRow)
null)
- :
parser.parseBatch(ByteBuffer.wrap(valueBytes));
- boolean isPersistRequired = false;
-
- final SequenceMetadata sequenceToUse = sequences
- .stream()
- .filter(sequenceMetadata ->
sequenceMetadata.canHandle(record))
- .findFirst()
- .orElse(null);
-
- if (sequenceToUse == null) {
- throw new ISE(
- "WTH?! cannot find any valid sequence for record with
partition [%d] and offset [%d]. Current sequences: %s",
- record.partition(),
- record.offset(),
- sequences
- );
- }
-
- for (InputRow row : rows) {
- if (row != null && task.withinMinMaxRecordTime(row)) {
- final AppenderatorDriverAddResult addResult = driver.add(
- row,
- sequenceToUse.getSequenceName(),
- committerSupplier,
- // skip segment lineage check as there will always be
one segment
- // for combination of sequence and segment granularity.
- // It is necessary to skip it as the task puts
messages polled from all the
- // assigned Kafka partitions into a single Druid
segment, thus ordering of
- // messages among replica tasks across assigned
partitions is not guaranteed
- // which may cause replica tasks to ask for segments
with different interval
- // in different order which might cause
SegmentAllocateAction to fail.
- true,
- // do not allow incremental persists to happen until
all the rows from this batch
- // of rows are indexed
- false
- );
-
- if (addResult.isOk()) {
- // If the number of rows in the segment exceeds the
threshold after adding a row,
- // move the segment out from the active segments of
BaseAppenderatorDriver to make a new segment.
- if (addResult.isPushRequired(tuningConfig) &&
!sequenceToUse.isCheckpointed()) {
- sequenceToCheckpoint = sequenceToUse;
- }
- isPersistRequired |= addResult.isPersistRequired();
- } else {
- // Failure to allocate segment puts determinism at risk,
bail out to be safe.
- // May want configurable behavior here at some point.
- // If we allow continuing, then consider blacklisting
the interval for a while to avoid constant checks.
- throw new ISE("Could not allocate segment for row with
timestamp[%s]", row.getTimestamp());
- }
-
- if (addResult.getParseException() != null) {
- handleParseException(addResult.getParseException(),
record);
- } else {
- rowIngestionMeters.incrementProcessed();
- }
- } else {
- rowIngestionMeters.incrementThrownAway();
- }
- }
- if (isPersistRequired) {
- Futures.addCallback(
- driver.persistAsync(committerSupplier.get()),
- new FutureCallback<Object>()
- {
- @Override
- public void onSuccess(@Nullable Object result)
- {
- log.info("Persist completed with metadata [%s]",
result);
- }
-
- @Override
- public void onFailure(Throwable t)
- {
- log.error("Persist failed, dying");
- backgroundThreadException = t;
- }
- }
- );
- }
- }
- catch (ParseException e) {
- handleParseException(e, record);
- }
-
- nextOffsets.put(record.partition(), record.offset() + 1);
- }
-
- if
(nextOffsets.get(record.partition()).equals(endOffsets.get(record.partition()))
- && assignment.remove(record.partition())) {
- log.info("Finished reading topic[%s], partition[%,d].",
record.topic(), record.partition());
- KafkaIndexTask.assignPartitions(consumer, topic, assignment);
- stillReading = !assignment.isEmpty();
- }
- }
-
- if (System.currentTimeMillis() > nextCheckpointTime) {
- sequenceToCheckpoint = sequences.get(sequences.size() - 1);
- }
-
- if (sequenceToCheckpoint != null && stillReading) {
- Preconditions.checkArgument(
- sequences.get(sequences.size() - 1)
- .getSequenceName()
- .equals(sequenceToCheckpoint.getSequenceName()),
- "Cannot checkpoint a sequence [%s] which is not the latest
one, sequences %s",
- sequenceToCheckpoint,
- sequences
- );
- requestPause();
- final CheckPointDataSourceMetadataAction checkpointAction = new
CheckPointDataSourceMetadataAction(
- task.getDataSource(),
- ioConfig.getTaskGroupId(),
- task.getIOConfig().getBaseSequenceName(),
- new KafkaDataSourceMetadata(new KafkaPartitions(topic,
sequenceToCheckpoint.getStartOffsets())),
- new KafkaDataSourceMetadata(new KafkaPartitions(topic,
nextOffsets))
- );
- if (!toolbox.getTaskActionClient().submit(checkpointAction)) {
- throw new ISE("Checkpoint request with offsets [%s] failed,
dying", nextOffsets);
- }
- }
- }
- ingestionState = IngestionState.COMPLETED;
- }
- catch (Exception e) {
- // (1) catch all exceptions while reading from kafka
- caughtExceptionInner = e;
- log.error(e, "Encountered exception in run() before persisting.");
- throw e;
- }
- finally {
- log.info("Persisting all pending data");
- try {
- driver.persist(committerSupplier.get()); // persist pending data
- }
- catch (Exception e) {
- if (caughtExceptionInner != null) {
- caughtExceptionInner.addSuppressed(e);
- } else {
- throw e;
- }
- }
- }
-
- synchronized (statusLock) {
- if (stopRequested.get() && !publishOnStop.get()) {
- throw new InterruptedException("Stopping without publishing");
- }
-
- status = Status.PUBLISHING;
- }
-
- for (SequenceMetadata sequenceMetadata : sequences) {
- if (!publishingSequences.contains(sequenceMetadata.getSequenceName()))
{
- // this is done to prevent checks in sequence specific commit
supplier from failing
- sequenceMetadata.setEndOffsets(nextOffsets);
- sequenceMetadata.updateAssignments(nextOffsets);
- publishingSequences.add(sequenceMetadata.getSequenceName());
- // persist already done in finally, so directly add to publishQueue
- publishAndRegisterHandoff(sequenceMetadata);
- }
- }
-
- if (backgroundThreadException != null) {
- throw new RuntimeException(backgroundThreadException);
- }
-
- // Wait for publish futures to complete.
- Futures.allAsList(publishWaitList).get();
-
- // Wait for handoff futures to complete.
- // Note that every publishing task (created by calling
AppenderatorDriver.publish()) has a corresponding
- // handoffFuture. handoffFuture can throw an exception if 1) the
corresponding publishFuture failed or 2) it
- // failed to persist sequences. It might also return null if handoff
failed, but was recoverable.
- // See publishAndRegisterHandoff() for details.
- List<SegmentsAndMetadata> handedOffList = Collections.emptyList();
- if (tuningConfig.getHandoffConditionTimeout() == 0) {
- handedOffList = Futures.allAsList(handOffWaitList).get();
- } else {
- try {
- handedOffList = Futures.allAsList(handOffWaitList)
-
.get(tuningConfig.getHandoffConditionTimeout(), TimeUnit.MILLISECONDS);
- }
- catch (TimeoutException e) {
- // Handoff timeout is not an indexing failure, but coordination
failure. We simply ignore timeout exception
- // here.
- log.makeAlert("Timed out after [%d] millis waiting for handoffs",
tuningConfig.getHandoffConditionTimeout())
- .addData("TaskId", task.getId())
- .emit();
- }
- }
-
- for (SegmentsAndMetadata handedOff : handedOffList) {
- log.info(
- "Handoff completed for segments[%s] with metadata[%s].",
- Joiner.on(", ").join(
-
handedOff.getSegments().stream().map(DataSegment::getIdentifier).collect(Collectors.toList())
- ),
- Preconditions.checkNotNull(handedOff.getCommitMetadata(),
"commitMetadata")
- );
- }
-
- appenderator.close();
- }
- catch (InterruptedException | RejectedExecutionException e) {
- // (2) catch InterruptedException and RejectedExecutionException thrown
for the whole ingestion steps including
- // the final publishing.
- caughtExceptionOuter = e;
- try {
- Futures.allAsList(publishWaitList).cancel(true);
- Futures.allAsList(handOffWaitList).cancel(true);
- if (appenderator != null) {
- appenderator.closeNow();
- }
- }
- catch (Exception e2) {
- e.addSuppressed(e2);
- }
-
- // handle the InterruptedException that gets wrapped in a
RejectedExecutionException
- if (e instanceof RejectedExecutionException
- && (e.getCause() == null || !(e.getCause() instanceof
InterruptedException))) {
- throw e;
- }
-
- // if we were interrupted because we were asked to stop, handle the
exception and return success, else rethrow
- if (!stopRequested.get()) {
- Thread.currentThread().interrupt();
- throw e;
- }
-
- log.info("The task was asked to stop before completing");
- }
- catch (Exception e) {
- // (3) catch all other exceptions thrown for the whole ingestion steps
including the final publishing.
- caughtExceptionOuter = e;
- try {
- Futures.allAsList(publishWaitList).cancel(true);
- Futures.allAsList(handOffWaitList).cancel(true);
- if (appenderator != null) {
- appenderator.closeNow();
- }
- }
- catch (Exception e2) {
- e.addSuppressed(e2);
- }
- throw e;
- }
- finally {
- try {
- if (driver != null) {
- driver.close();
- }
- if (chatHandlerProvider.isPresent()) {
- chatHandlerProvider.get().unregister(task.getId());
- }
-
- toolbox.getDruidNodeAnnouncer().unannounce(discoveryDruidNode);
- toolbox.getDataSegmentServerAnnouncer().unannounce();
- }
- catch (Exception e) {
- if (caughtExceptionOuter != null) {
- caughtExceptionOuter.addSuppressed(e);
- } else {
- throw e;
- }
- }
- }
-
- toolbox.getTaskReportFileWriter().write(getTaskCompletionReports(null));
- return TaskStatus.success(task.getId());
- }
-
- private void checkPublishAndHandoffFailure() throws ExecutionException,
InterruptedException
+ protected Long getSequenceNumberToStoreAfterRead(@NotNull Long
sequenceNumber)
{
- // Check if any publishFuture failed.
- final List<ListenableFuture<SegmentsAndMetadata>> publishFinished =
publishWaitList
- .stream()
- .filter(Future::isDone)
- .collect(Collectors.toList());
-
- for (ListenableFuture<SegmentsAndMetadata> publishFuture :
publishFinished) {
- // If publishFuture failed, the below line will throw an exception and
catched by (1), and then (2) or (3).
- publishFuture.get();
- }
-
- publishWaitList.removeAll(publishFinished);
-
- // Check if any handoffFuture failed.
- final List<ListenableFuture<SegmentsAndMetadata>> handoffFinished =
handOffWaitList
- .stream()
- .filter(Future::isDone)
- .collect(Collectors.toList());
-
- for (ListenableFuture<SegmentsAndMetadata> handoffFuture :
handoffFinished) {
- // If handoffFuture failed, the below line will throw an exception and
catched by (1), and then (2) or (3).
- handoffFuture.get();
- }
-
- handOffWaitList.removeAll(handoffFinished);
+ return sequenceNumber + 1;
}
- private void publishAndRegisterHandoff(SequenceMetadata sequenceMetadata)
- {
- log.info("Publishing segments for sequence [%s]", sequenceMetadata);
-
- final ListenableFuture<SegmentsAndMetadata> publishFuture =
Futures.transform(
- driver.publish(
- sequenceMetadata.createPublisher(toolbox,
ioConfig.isUseTransaction()),
- sequenceMetadata.getCommitterSupplier(topic,
lastPersistedOffsets).get(),
- Collections.singletonList(sequenceMetadata.getSequenceName())
- ),
- (Function<SegmentsAndMetadata, SegmentsAndMetadata>)
publishedSegmentsAndMetadata -> {
- if (publishedSegmentsAndMetadata == null) {
- throw new ISE(
- "Transaction failure publishing segments for sequence [%s]",
- sequenceMetadata
- );
- } else {
- return publishedSegmentsAndMetadata;
- }
- }
- );
- publishWaitList.add(publishFuture);
-
- // Create a handoffFuture for every publishFuture. The created
handoffFuture must fail if publishFuture fails.
- final SettableFuture<SegmentsAndMetadata> handoffFuture =
SettableFuture.create();
- handOffWaitList.add(handoffFuture);
-
- Futures.addCallback(
- publishFuture,
- new FutureCallback<SegmentsAndMetadata>()
- {
- @Override
- public void onSuccess(SegmentsAndMetadata
publishedSegmentsAndMetadata)
- {
- log.info(
- "Published segments[%s] with metadata[%s].",
- publishedSegmentsAndMetadata.getSegments()
- .stream()
- .map(DataSegment::getIdentifier)
- .collect(Collectors.toList()),
-
Preconditions.checkNotNull(publishedSegmentsAndMetadata.getCommitMetadata(),
"commitMetadata")
- );
-
- sequences.remove(sequenceMetadata);
- publishingSequences.remove(sequenceMetadata.getSequenceName());
- try {
- persistSequences();
- }
- catch (IOException e) {
- log.error(e, "Unable to persist state, dying");
- handoffFuture.setException(e);
- throw new RuntimeException(e);
- }
-
- Futures.transform(
- driver.registerHandoff(publishedSegmentsAndMetadata),
- new Function<SegmentsAndMetadata, Void>()
- {
- @Nullable
- @Override
- public Void apply(@Nullable SegmentsAndMetadata
handoffSegmentsAndMetadata)
- {
- if (handoffSegmentsAndMetadata == null) {
- log.warn(
- "Failed to handoff segments[%s]",
- publishedSegmentsAndMetadata.getSegments()
- .stream()
-
.map(DataSegment::getIdentifier)
-
.collect(Collectors.toList())
- );
- }
- handoffFuture.set(handoffSegmentsAndMetadata);
- return null;
- }
- }
- );
- }
-
- @Override
- public void onFailure(Throwable t)
- {
- log.error(t, "Error while publishing segments for sequence[%s]",
sequenceMetadata);
- handoffFuture.setException(t);
- }
- }
- );
- }
-
- private static File getSequencesPersistFile(TaskToolbox toolbox)
- {
- return new File(toolbox.getPersistDir(), "sequences.json");
- }
-
- private boolean restoreSequences() throws IOException
- {
- final File sequencesPersistFile = getSequencesPersistFile(toolbox);
- if (sequencesPersistFile.exists()) {
- sequences = new CopyOnWriteArrayList<>(
- toolbox.getObjectMapper().<List<SequenceMetadata>>readValue(
- sequencesPersistFile,
- new TypeReference<List<SequenceMetadata>>()
- {
- }
- )
- );
- return true;
- } else {
- return false;
- }
- }
-
- private synchronized void persistSequences() throws IOException
- {
- log.info("Persisting Sequences Metadata [%s]", sequences);
- toolbox.getObjectMapper().writerWithType(
- new TypeReference<List<SequenceMetadata>>()
- {
- }
- ).writeValue(getSequencesPersistFile(toolbox), sequences);
- }
-
- private Map<String, TaskReport> getTaskCompletionReports(@Nullable String
errorMsg)
- {
- return TaskReport.buildTaskReports(
- new IngestionStatsAndErrorsTaskReport(
- task.getId(),
- new IngestionStatsAndErrorsTaskReportData(
- ingestionState,
- getTaskCompletionUnparseableEvents(),
- getTaskCompletionRowStats(),
- errorMsg
- )
- )
- );
- }
-
- private Map<String, Object> getTaskCompletionUnparseableEvents()
- {
- Map<String, Object> unparseableEventsMap = new HashMap<>();
- List<String> buildSegmentsParseExceptionMessages =
IndexTaskUtils.getMessagesFromSavedParseExceptions(
- savedParseExceptions
- );
- if (buildSegmentsParseExceptionMessages != null) {
- unparseableEventsMap.put(RowIngestionMeters.BUILD_SEGMENTS,
buildSegmentsParseExceptionMessages);
- }
- return unparseableEventsMap;
- }
-
- private Map<String, Object> getTaskCompletionRowStats()
- {
- Map<String, Object> metrics = new HashMap<>();
- metrics.put(
- RowIngestionMeters.BUILD_SEGMENTS,
- rowIngestionMeters.getTotals()
- );
- return metrics;
- }
-
- private void maybePersistAndPublishSequences(Supplier<Committer>
committerSupplier)
- throws InterruptedException
- {
- for (SequenceMetadata sequenceMetadata : sequences) {
- sequenceMetadata.updateAssignments(nextOffsets);
- if (!sequenceMetadata.isOpen() &&
!publishingSequences.contains(sequenceMetadata.getSequenceName())) {
- publishingSequences.add(sequenceMetadata.getSequenceName());
- try {
- Object result = driver.persist(committerSupplier.get());
- log.info(
- "Persist completed with results: [%s], adding sequence [%s] to
publish queue",
- result,
- sequenceMetadata
- );
- publishAndRegisterHandoff(sequenceMetadata);
- }
- catch (InterruptedException e) {
- log.warn("Interrupted while persisting sequence [%s]",
sequenceMetadata);
- throw e;
- }
- }
- }
- }
-
- private Set<Integer> assignPartitionsAndSeekToNext(KafkaConsumer consumer,
String topic)
- {
- // Initialize consumer assignment.
- final Set<Integer> assignment = new HashSet<>();
- for (Map.Entry<Integer, Long> entry : nextOffsets.entrySet()) {
- final long endOffset = endOffsets.get(entry.getKey());
- if (entry.getValue() < endOffset) {
- assignment.add(entry.getKey());
- } else if (entry.getValue() == endOffset) {
- log.info("Finished reading partition[%d].", entry.getKey());
- } else {
- throw new ISE(
- "WTF?! Cannot start from offset[%,d] > endOffset[%,d]",
- entry.getValue(),
- endOffset
- );
- }
+ @Nonnull
+ @Override
+ protected List<OrderedPartitionableRecord<Integer, Long>> getRecords(
+ RecordSupplier<Integer, Long> recordSupplier,
+ TaskToolbox toolbox
+ ) throws Exception
+ {
+ // The retrying business is because the KafkaConsumer throws
OffsetOutOfRangeException if the seeked-to
Review comment:
I know that this comment is just copied, but now it's not easy to understand
where the retrying business is. Would you improve this comment?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
