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_r241923675
##########
File path:
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java
##########
@@ -2375,239 +343,48 @@ private Runnable emitLag()
};
}
- private void updateCurrentOffsets() throws InterruptedException,
ExecutionException, TimeoutException
- {
- final List<ListenableFuture<Void>> futures = Stream.concat(
- taskGroups.values().stream().flatMap(taskGroup ->
taskGroup.tasks.entrySet().stream()),
- pendingCompletionTaskGroups.values()
- .stream()
- .flatMap(List::stream)
- .flatMap(taskGroup ->
taskGroup.tasks.entrySet().stream())
- ).map(
- task -> Futures.transform(
- taskClient.getCurrentOffsetsAsync(task.getKey(), false),
- (Function<Map<Integer, Long>, Void>) (currentOffsets) -> {
-
- if (currentOffsets != null && !currentOffsets.isEmpty()) {
- task.getValue().currentOffsets = currentOffsets;
- }
-
- return null;
- }
- )
- ).collect(Collectors.toList());
-
- Futures.successfulAsList(futures).get(futureTimeoutInSeconds,
TimeUnit.SECONDS);
- }
-
- @VisibleForTesting
- Runnable updateCurrentAndLatestOffsets()
- {
- return () -> {
- try {
- updateCurrentOffsets();
- updateLatestOffsetsFromKafka();
- offsetsLastUpdated = DateTimes.nowUtc();
- }
- catch (Exception e) {
- log.warn(e, "Exception while getting current/latest offsets");
- }
- };
- }
-
- /**
- * Collect row ingestion stats from all tasks managed by this supervisor.
- *
- * @return A map of groupId->taskId->task row stats
- *
- * @throws InterruptedException
- * @throws ExecutionException
- * @throws TimeoutException
- */
- private Map<String, Map<String, Object>> getCurrentTotalStats()
- throws InterruptedException, ExecutionException, TimeoutException
+ @Override
+ protected Long getNotSetMarker()
{
- Map<String, Map<String, Object>> allStats = new HashMap<>();
- final List<ListenableFuture<StatsFromTaskResult>> futures = new
ArrayList<>();
- final List<Pair<Integer, String>> groupAndTaskIds = new ArrayList<>();
-
- for (int groupId : taskGroups.keySet()) {
- TaskGroup group = taskGroups.get(groupId);
- for (String taskId : group.taskIds()) {
- futures.add(
- Futures.transform(
- taskClient.getMovingAveragesAsync(taskId),
- (Function<Map<String, Object>, StatsFromTaskResult>)
(currentStats) -> {
- return new StatsFromTaskResult(
- groupId,
- taskId,
- currentStats
- );
- }
- )
- );
- groupAndTaskIds.add(new Pair<>(groupId, taskId));
- }
- }
-
- for (int groupId : pendingCompletionTaskGroups.keySet()) {
- List<TaskGroup> pendingGroups = pendingCompletionTaskGroups.get(groupId);
- for (TaskGroup pendingGroup : pendingGroups) {
- for (String taskId : pendingGroup.taskIds()) {
- futures.add(
- Futures.transform(
- taskClient.getMovingAveragesAsync(taskId),
- (Function<Map<String, Object>, StatsFromTaskResult>)
(currentStats) -> new StatsFromTaskResult(
- groupId,
- taskId,
- currentStats
- )
- )
- );
- groupAndTaskIds.add(new Pair<>(groupId, taskId));
- }
- }
- }
-
- List<StatsFromTaskResult> results =
Futures.successfulAsList(futures).get(futureTimeoutInSeconds, TimeUnit.SECONDS);
- for (int i = 0; i < results.size(); i++) {
- StatsFromTaskResult result = results.get(i);
- if (result != null) {
- Map<String, Object> groupMap =
allStats.computeIfAbsent(result.getGroupId(), k -> new HashMap<>());
- groupMap.put(result.getTaskId(), result.getStats());
- } else {
- Pair<Integer, String> groupAndTaskId = groupAndTaskIds.get(i);
- log.error("Failed to get stats for group[%d]-task[%s]",
groupAndTaskId.lhs, groupAndTaskId.rhs);
- }
- }
-
- return allStats;
+ return NOT_SET;
}
- @VisibleForTesting
- void addTaskGroupToActivelyReadingTaskGroup(
- int taskGroupId,
- ImmutableMap<Integer, Long> partitionOffsets,
- Optional<DateTime> minMsgTime,
- Optional<DateTime> maxMsgTime,
- Set<String> tasks
- )
+ @Override
+ protected Long getEndOfPartitionMarker()
{
- TaskGroup group = new TaskGroup(
- taskGroupId,
- partitionOffsets,
- minMsgTime,
- maxMsgTime
- );
- group.tasks.putAll(tasks.stream().collect(Collectors.toMap(x -> x, x ->
new TaskData())));
- if (taskGroups.putIfAbsent(taskGroupId, group) != null) {
- throw new ISE(
- "trying to add taskGroup with ID [%s] to actively reading task
groups, but group already exists.",
- taskGroupId
- );
- }
+ return END_OF_PARTITION;
}
- @VisibleForTesting
- void addTaskGroupToPendingCompletionTaskGroup(
- int taskGroupId,
- ImmutableMap<Integer, Long> partitionOffsets,
- Optional<DateTime> minMsgTime,
- Optional<DateTime> maxMsgTime,
- Set<String> tasks
+ @Override
+ protected void updateLatestSequenceFromStream(
+ RecordSupplier<Integer, Long> recordSupplier,
+ Set<StreamPartition<Integer>> partitions
)
{
- TaskGroup group = new TaskGroup(
- taskGroupId,
- partitionOffsets,
- minMsgTime,
- maxMsgTime
- );
- group.tasks.putAll(tasks.stream().collect(Collectors.toMap(x -> x, x ->
new TaskData())));
- pendingCompletionTaskGroups.computeIfAbsent(taskGroupId, x -> new
CopyOnWriteArrayList<>())
- .add(group);
- }
-
- @VisibleForTesting
- @Nullable
- TaskGroup removeTaskGroup(int taskGroupId)
- {
- return taskGroups.remove(taskGroupId);
- }
-
- @VisibleForTesting
- void moveTaskGroupToPendingCompletion(int taskGroupId)
- {
- final TaskGroup taskGroup = taskGroups.remove(taskGroupId);
- if (taskGroup != null) {
- pendingCompletionTaskGroups.computeIfAbsent(taskGroupId, k -> new
CopyOnWriteArrayList<>()).add(taskGroup);
- }
- }
-
- @VisibleForTesting
- int getNoticesQueueSize()
- {
- return notices.size();
- }
-
- private static class StatsFromTaskResult
- {
- private final String groupId;
- private final String taskId;
- private final Map<String, Object> stats;
-
- public StatsFromTaskResult(
- int groupId,
- String taskId,
- Map<String, Object> stats
- )
- {
- this.groupId = String.valueOf(groupId);
- this.taskId = taskId;
- this.stats = stats;
- }
-
- public String getGroupId()
- {
- return groupId;
- }
-
- public String getTaskId()
- {
- return taskId;
- }
-
- public Map<String, Object> getStats()
- {
- return stats;
- }
- }
-
- // exposed for testing for visibility into initialization state
- @VisibleForTesting
- public boolean isStarted()
- {
- return started;
+ latestSequenceFromStream = partitions.stream()
+ .collect(Collectors.toMap(
+ StreamPartition::getPartitionId,
+ recordSupplier::getPosition
+ ));
}
- // exposed for testing for visibility into initialization state
- @VisibleForTesting
- public boolean isLifecycleStarted()
+ @Override
+ protected String baseTaskName()
{
- return lifecycleStarted;
+ return "index_kafka";
}
- // exposed for testing for visibility into initialization state
+ @Override
@VisibleForTesting
- public int getInitRetryCounter()
+ public KafkaSupervisorIOConfig getIoConfig()
{
- return initRetryCounter;
+ return spec.getIoConfig();
}
- // exposed for testing to allow "bootstrap.servers" to be changed after
supervisor is created
+ @Override
@VisibleForTesting
- public KafkaSupervisorIOConfig getIoConfig()
+ protected void tryInit()
{
- return ioConfig;
+ super.tryInit();
Review comment:
Please remove this.
----------------------------------------------------------------
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]
