You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "vvivekiyer (via GitHub)" <gi...@apache.org> on 2024/12/04 11:02:50 UTC
[PR] RefreshSegmentTask - Support for scheduling adhoc tasks [pinot]
vvivekiyer opened a new pull request, #14596:
URL: https://github.com/apache/pinot/pull/14596
PR https://github.com/apache/pinot/pull/14300/files added support for automatic segment refreshes.
This PR adds support for scheduling ad-hoc RefreshSegment tasks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] RefreshSegmentTask - Support for scheduling adhoc tasks [pinot]
Posted by "vvivekiyer (via GitHub)" <gi...@apache.org>.
vvivekiyer commented on code in PR #14596:
URL: https://github.com/apache/pinot/pull/14596#discussion_r1870341390
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/refreshsegment/RefreshSegmentTaskGenerator.java:
##########
@@ -54,84 +54,99 @@ public String getTaskType() {
@Override
public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
- String taskType = RefreshSegmentTask.TASK_TYPE;
List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
- PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
-
- int tableNumTasks = 0;
-
for (TableConfig tableConfig : tableConfigs) {
- String tableNameWithType = tableConfig.getTableName();
- LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
-
// Get the task configs for the table. This is used to restrict the maximum number of allowed tasks per table at
// any given point.
Map<String, String> taskConfigs;
TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
if (tableTaskConfig == null) {
- LOGGER.warn("Failed to find task config for table: {}", tableNameWithType);
+ LOGGER.warn("Failed to find task config for table: {}", tableConfig.getTableName());
continue;
}
taskConfigs = tableTaskConfig.getConfigsForTaskType(RefreshSegmentTask.TASK_TYPE);
- Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableNameWithType);
- int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
- if (tableMaxNumTasksConfig != null) {
- try {
- tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
- } catch (Exception e) {
- tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
- }
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s",
+ tableConfig.getTableName());
+
+ pinotTaskConfigs.addAll(generateTasksForTable(tableConfig, taskConfigs));
+ }
+
+ return pinotTaskConfigs;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(TableConfig tableConfig, Map<String, String> taskConfigs)
+ throws Exception {
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableConfig.getTableName());
+ return generateTasksForTable(tableConfig, taskConfigs);
+ }
+
+ private List<PinotTaskConfig> generateTasksForTable(TableConfig tableConfig, Map<String, String> taskConfigs) {
+ String taskType = RefreshSegmentTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+ PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
+
+ String tableNameWithType = tableConfig.getTableName();
+ LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
+
+
+ int tableNumTasks = 0;
+ int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
Review Comment:
Good point. Can you please add these thoughts to this issue - https://github.com/apache/pinot/issues/14483 where we can do it as a followup?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] RefreshSegmentTask - Support for scheduling adhoc tasks [pinot]
Posted by "tibrewalpratik17 (via GitHub)" <gi...@apache.org>.
tibrewalpratik17 commented on code in PR #14596:
URL: https://github.com/apache/pinot/pull/14596#discussion_r1870035109
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/refreshsegment/RefreshSegmentTaskGenerator.java:
##########
@@ -54,84 +54,99 @@ public String getTaskType() {
@Override
public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
- String taskType = RefreshSegmentTask.TASK_TYPE;
List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
- PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
-
- int tableNumTasks = 0;
-
for (TableConfig tableConfig : tableConfigs) {
- String tableNameWithType = tableConfig.getTableName();
- LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
-
// Get the task configs for the table. This is used to restrict the maximum number of allowed tasks per table at
// any given point.
Map<String, String> taskConfigs;
TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
if (tableTaskConfig == null) {
- LOGGER.warn("Failed to find task config for table: {}", tableNameWithType);
+ LOGGER.warn("Failed to find task config for table: {}", tableConfig.getTableName());
continue;
}
taskConfigs = tableTaskConfig.getConfigsForTaskType(RefreshSegmentTask.TASK_TYPE);
- Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableNameWithType);
- int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
- if (tableMaxNumTasksConfig != null) {
- try {
- tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
- } catch (Exception e) {
- tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
- }
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s",
+ tableConfig.getTableName());
+
+ pinotTaskConfigs.addAll(generateTasksForTable(tableConfig, taskConfigs));
+ }
+
+ return pinotTaskConfigs;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(TableConfig tableConfig, Map<String, String> taskConfigs)
+ throws Exception {
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableConfig.getTableName());
+ return generateTasksForTable(tableConfig, taskConfigs);
+ }
+
+ private List<PinotTaskConfig> generateTasksForTable(TableConfig tableConfig, Map<String, String> taskConfigs) {
+ String taskType = RefreshSegmentTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+ PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
+
+ String tableNameWithType = tableConfig.getTableName();
+ LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
+
+
+ int tableNumTasks = 0;
+ int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
Review Comment:
not introduced in this PR but any rationale for keeping RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE default value as 20.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] RefreshSegmentTask - Support for scheduling adhoc tasks [pinot]
Posted by "vvivekiyer (via GitHub)" <gi...@apache.org>.
vvivekiyer commented on code in PR #14596:
URL: https://github.com/apache/pinot/pull/14596#discussion_r1870196514
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/refreshsegment/RefreshSegmentTaskGenerator.java:
##########
@@ -54,84 +54,99 @@ public String getTaskType() {
@Override
public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
- String taskType = RefreshSegmentTask.TASK_TYPE;
List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
- PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
-
- int tableNumTasks = 0;
-
for (TableConfig tableConfig : tableConfigs) {
- String tableNameWithType = tableConfig.getTableName();
- LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
-
// Get the task configs for the table. This is used to restrict the maximum number of allowed tasks per table at
// any given point.
Map<String, String> taskConfigs;
TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
if (tableTaskConfig == null) {
- LOGGER.warn("Failed to find task config for table: {}", tableNameWithType);
+ LOGGER.warn("Failed to find task config for table: {}", tableConfig.getTableName());
continue;
}
taskConfigs = tableTaskConfig.getConfigsForTaskType(RefreshSegmentTask.TASK_TYPE);
- Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableNameWithType);
- int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
- if (tableMaxNumTasksConfig != null) {
- try {
- tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
- } catch (Exception e) {
- tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
- }
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s",
+ tableConfig.getTableName());
+
+ pinotTaskConfigs.addAll(generateTasksForTable(tableConfig, taskConfigs));
+ }
+
+ return pinotTaskConfigs;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(TableConfig tableConfig, Map<String, String> taskConfigs)
+ throws Exception {
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableConfig.getTableName());
+ return generateTasksForTable(tableConfig, taskConfigs);
+ }
+
+ private List<PinotTaskConfig> generateTasksForTable(TableConfig tableConfig, Map<String, String> taskConfigs) {
+ String taskType = RefreshSegmentTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+ PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
+
+ String tableNameWithType = tableConfig.getTableName();
+ LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
+
+
+ int tableNumTasks = 0;
+ int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
Review Comment:
No particular reason - chose a low value to make sure we don't put stress on deployments that have few minions. This can be overriden with table config.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] RefreshSegmentTask - Support for scheduling adhoc tasks [pinot]
Posted by "jasperjiaguo (via GitHub)" <gi...@apache.org>.
jasperjiaguo merged PR #14596:
URL: https://github.com/apache/pinot/pull/14596
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] RefreshSegmentTask - Support for scheduling adhoc tasks [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #14596:
URL: https://github.com/apache/pinot/pull/14596#issuecomment-2517096043
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/14596?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 0.00%. Comparing base [(`59551e4`)](https://app.codecov.io/gh/apache/pinot/commit/59551e45224f1535c4863fd577622b37366ccc97?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`f19a850`)](https://app.codecov.io/gh/apache/pinot/commit/f19a8505ec899402a20b845e5e34a39477c894be?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1424 commits behind head on master.
> :exclamation: There is a different number of reports uploaded between BASE (59551e4) and HEAD (f19a850). Click for more details.
>
> <details><summary>HEAD has 53 uploads less than BASE</summary>
>
>| Flag | BASE (59551e4) | HEAD (f19a850) |
>|------|------|------|
>|integration|7|1|
>|integration2|3|1|
>|temurin|12|1|
>|java-21|7|1|
>|skip-bytebuffers-true|3|1|
>|skip-bytebuffers-false|7|0|
>|unittests|5|0|
>|unittests1|2|0|
>|java-11|5|0|
>|unittests2|3|0|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #14596 +/- ##
=============================================
- Coverage 61.75% 0.00% -61.76%
=============================================
Files 2436 3 -2433
Lines 133233 6 -133227
Branches 20636 0 -20636
=============================================
- Hits 82274 0 -82274
+ Misses 44911 6 -44905
+ Partials 6048 0 -6048
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-0.01%)` | :arrow_down: |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-61.63%)` | :arrow_down: |
| [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [skip-bytebuffers-true](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-27.73%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-61.76%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/14596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/14596?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] RefreshSegmentTask - Support for scheduling adhoc tasks [pinot]
Posted by "tibrewalpratik17 (via GitHub)" <gi...@apache.org>.
tibrewalpratik17 commented on code in PR #14596:
URL: https://github.com/apache/pinot/pull/14596#discussion_r1870031425
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/refreshsegment/RefreshSegmentTaskGenerator.java:
##########
@@ -54,84 +54,99 @@ public String getTaskType() {
@Override
public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
- String taskType = RefreshSegmentTask.TASK_TYPE;
List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
- PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
-
- int tableNumTasks = 0;
-
for (TableConfig tableConfig : tableConfigs) {
- String tableNameWithType = tableConfig.getTableName();
- LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
-
// Get the task configs for the table. This is used to restrict the maximum number of allowed tasks per table at
// any given point.
Map<String, String> taskConfigs;
TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
if (tableTaskConfig == null) {
- LOGGER.warn("Failed to find task config for table: {}", tableNameWithType);
+ LOGGER.warn("Failed to find task config for table: {}", tableConfig.getTableName());
continue;
}
taskConfigs = tableTaskConfig.getConfigsForTaskType(RefreshSegmentTask.TASK_TYPE);
- Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableNameWithType);
- int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
- if (tableMaxNumTasksConfig != null) {
- try {
- tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
- } catch (Exception e) {
- tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
- }
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s",
+ tableConfig.getTableName());
+
+ pinotTaskConfigs.addAll(generateTasksForTable(tableConfig, taskConfigs));
+ }
+
+ return pinotTaskConfigs;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(TableConfig tableConfig, Map<String, String> taskConfigs)
+ throws Exception {
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableConfig.getTableName());
+ return generateTasksForTable(tableConfig, taskConfigs);
+ }
+
+ private List<PinotTaskConfig> generateTasksForTable(TableConfig tableConfig, Map<String, String> taskConfigs) {
+ String taskType = RefreshSegmentTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+ PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
+
+ String tableNameWithType = tableConfig.getTableName();
+ LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
+
+
+ int tableNumTasks = 0;
+ int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
+ String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
+ if (tableMaxNumTasksConfig != null) {
+ try {
+ tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
+ } catch (Exception e) {
+ tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
+ LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
+ }
+ }
Review Comment:
Ca
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/refreshsegment/RefreshSegmentTaskGenerator.java:
##########
@@ -54,84 +54,99 @@ public String getTaskType() {
@Override
public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
- String taskType = RefreshSegmentTask.TASK_TYPE;
List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
- PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
-
- int tableNumTasks = 0;
-
for (TableConfig tableConfig : tableConfigs) {
- String tableNameWithType = tableConfig.getTableName();
- LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
-
// Get the task configs for the table. This is used to restrict the maximum number of allowed tasks per table at
// any given point.
Map<String, String> taskConfigs;
TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
if (tableTaskConfig == null) {
- LOGGER.warn("Failed to find task config for table: {}", tableNameWithType);
+ LOGGER.warn("Failed to find task config for table: {}", tableConfig.getTableName());
continue;
}
taskConfigs = tableTaskConfig.getConfigsForTaskType(RefreshSegmentTask.TASK_TYPE);
- Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableNameWithType);
- int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
- if (tableMaxNumTasksConfig != null) {
- try {
- tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
- } catch (Exception e) {
- tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
- }
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s",
+ tableConfig.getTableName());
+
+ pinotTaskConfigs.addAll(generateTasksForTable(tableConfig, taskConfigs));
+ }
+
+ return pinotTaskConfigs;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(TableConfig tableConfig, Map<String, String> taskConfigs)
+ throws Exception {
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableConfig.getTableName());
+ return generateTasksForTable(tableConfig, taskConfigs);
+ }
+
+ private List<PinotTaskConfig> generateTasksForTable(TableConfig tableConfig, Map<String, String> taskConfigs) {
+ String taskType = RefreshSegmentTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+ PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
+
+ String tableNameWithType = tableConfig.getTableName();
+ LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
+
+
+ int tableNumTasks = 0;
+ int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
+ String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
+ if (tableMaxNumTasksConfig != null) {
+ try {
+ tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
+ } catch (Exception e) {
+ tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
+ LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
+ }
+ }
Review Comment:
Ca
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] RefreshSegmentTask - Support for scheduling adhoc tasks [pinot]
Posted by "sajjad-moradi (via GitHub)" <gi...@apache.org>.
sajjad-moradi commented on code in PR #14596:
URL: https://github.com/apache/pinot/pull/14596#discussion_r1870299501
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/refreshsegment/RefreshSegmentTaskGenerator.java:
##########
@@ -54,84 +54,99 @@ public String getTaskType() {
@Override
public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
- String taskType = RefreshSegmentTask.TASK_TYPE;
List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
- PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
-
- int tableNumTasks = 0;
-
for (TableConfig tableConfig : tableConfigs) {
- String tableNameWithType = tableConfig.getTableName();
- LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
-
// Get the task configs for the table. This is used to restrict the maximum number of allowed tasks per table at
// any given point.
Map<String, String> taskConfigs;
TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
if (tableTaskConfig == null) {
- LOGGER.warn("Failed to find task config for table: {}", tableNameWithType);
+ LOGGER.warn("Failed to find task config for table: {}", tableConfig.getTableName());
continue;
}
taskConfigs = tableTaskConfig.getConfigsForTaskType(RefreshSegmentTask.TASK_TYPE);
- Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableNameWithType);
- int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
- if (tableMaxNumTasksConfig != null) {
- try {
- tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
- } catch (Exception e) {
- tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
- }
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s",
+ tableConfig.getTableName());
Review Comment:
nit: this precondition is repeated in both `generatedTasks` methods. It can be moved to `generatedTaskForTable` method to avoid duplication.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] RefreshSegmentTask - Support for scheduling adhoc tasks [pinot]
Posted by "tibrewalpratik17 (via GitHub)" <gi...@apache.org>.
tibrewalpratik17 commented on code in PR #14596:
URL: https://github.com/apache/pinot/pull/14596#discussion_r1870200378
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/refreshsegment/RefreshSegmentTaskGenerator.java:
##########
@@ -54,84 +54,99 @@ public String getTaskType() {
@Override
public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
- String taskType = RefreshSegmentTask.TASK_TYPE;
List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
- PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
-
- int tableNumTasks = 0;
-
for (TableConfig tableConfig : tableConfigs) {
- String tableNameWithType = tableConfig.getTableName();
- LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
-
// Get the task configs for the table. This is used to restrict the maximum number of allowed tasks per table at
// any given point.
Map<String, String> taskConfigs;
TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
if (tableTaskConfig == null) {
- LOGGER.warn("Failed to find task config for table: {}", tableNameWithType);
+ LOGGER.warn("Failed to find task config for table: {}", tableConfig.getTableName());
continue;
}
taskConfigs = tableTaskConfig.getConfigsForTaskType(RefreshSegmentTask.TASK_TYPE);
- Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableNameWithType);
- int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
- if (tableMaxNumTasksConfig != null) {
- try {
- tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
- } catch (Exception e) {
- tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
- }
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s",
+ tableConfig.getTableName());
+
+ pinotTaskConfigs.addAll(generateTasksForTable(tableConfig, taskConfigs));
+ }
+
+ return pinotTaskConfigs;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(TableConfig tableConfig, Map<String, String> taskConfigs)
+ throws Exception {
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableConfig.getTableName());
+ return generateTasksForTable(tableConfig, taskConfigs);
+ }
+
+ private List<PinotTaskConfig> generateTasksForTable(TableConfig tableConfig, Map<String, String> taskConfigs) {
+ String taskType = RefreshSegmentTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+ PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
+
+ String tableNameWithType = tableConfig.getTableName();
+ LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
+
+
+ int tableNumTasks = 0;
+ int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
Review Comment:
Hmm one improvement can be to keep it equal to number of partitions in a table and process one segment per partition (or instance) by default. This way the load gets normalized across hosts. We can do this as a follow-up too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] RefreshSegmentTask - Support for scheduling adhoc tasks [pinot]
Posted by "tibrewalpratik17 (via GitHub)" <gi...@apache.org>.
tibrewalpratik17 commented on code in PR #14596:
URL: https://github.com/apache/pinot/pull/14596#discussion_r1870037061
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/refreshsegment/RefreshSegmentTaskGenerator.java:
##########
@@ -54,84 +54,99 @@ public String getTaskType() {
@Override
public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
- String taskType = RefreshSegmentTask.TASK_TYPE;
List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
- PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
-
- int tableNumTasks = 0;
-
for (TableConfig tableConfig : tableConfigs) {
- String tableNameWithType = tableConfig.getTableName();
- LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
-
// Get the task configs for the table. This is used to restrict the maximum number of allowed tasks per table at
// any given point.
Map<String, String> taskConfigs;
TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
if (tableTaskConfig == null) {
- LOGGER.warn("Failed to find task config for table: {}", tableNameWithType);
+ LOGGER.warn("Failed to find task config for table: {}", tableConfig.getTableName());
continue;
}
taskConfigs = tableTaskConfig.getConfigsForTaskType(RefreshSegmentTask.TASK_TYPE);
- Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableNameWithType);
- int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
- if (tableMaxNumTasksConfig != null) {
- try {
- tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
- } catch (Exception e) {
- tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
- }
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s",
+ tableConfig.getTableName());
+
+ pinotTaskConfigs.addAll(generateTasksForTable(tableConfig, taskConfigs));
+ }
+
+ return pinotTaskConfigs;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(TableConfig tableConfig, Map<String, String> taskConfigs)
+ throws Exception {
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableConfig.getTableName());
+ return generateTasksForTable(tableConfig, taskConfigs);
+ }
+
+ private List<PinotTaskConfig> generateTasksForTable(TableConfig tableConfig, Map<String, String> taskConfigs) {
+ String taskType = RefreshSegmentTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+ PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
+
+ String tableNameWithType = tableConfig.getTableName();
+ LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
+
+
+ int tableNumTasks = 0;
+ int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
+ String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
+ if (tableMaxNumTasksConfig != null) {
+ try {
+ tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
+ } catch (Exception e) {
+ tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
+ LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
+ }
+ }
+
+ // Get info about table and schema.
+ Stat tableStat = pinotHelixResourceManager.getTableStat(tableNameWithType);
+ Schema schema = pinotHelixResourceManager.getSchemaForTableConfig(tableConfig);
+ Stat schemaStat = pinotHelixResourceManager.getSchemaStat(schema.getSchemaName());
+
+ // Get the running segments for a table.
+ Set<Segment> runningSegments =
+ TaskGeneratorUtils.getRunningSegments(RefreshSegmentTask.TASK_TYPE, _clusterInfoAccessor);
+
+ // Make a single ZK call to get the segments.
+ List<SegmentZKMetadata> allSegments = _clusterInfoAccessor.getSegmentsZKMetadata(tableNameWithType);
+
+ for (SegmentZKMetadata segmentZKMetadata : allSegments) {
+ // Skip if we have reached the maximum number of permissible tasks per iteration.
+ if (tableNumTasks >= tableMaxNumTasks) {
+ break;
}
- // Get info about table and schema.
- Stat tableStat = pinotHelixResourceManager.getTableStat(tableNameWithType);
- Schema schema = pinotHelixResourceManager.getSchemaForTableConfig(tableConfig);
- Stat schemaStat = pinotHelixResourceManager.getSchemaStat(schema.getSchemaName());
-
- // Get the running segments for a table.
- Set<Segment> runningSegments =
- TaskGeneratorUtils.getRunningSegments(RefreshSegmentTask.TASK_TYPE, _clusterInfoAccessor);
-
- // Make a single ZK call to get the segments.
- List<SegmentZKMetadata> allSegments = _clusterInfoAccessor.getSegmentsZKMetadata(tableNameWithType);
-
- for (SegmentZKMetadata segmentZKMetadata : allSegments) {
- // Skip if we have reached the maximum number of permissible tasks per iteration.
- if (tableNumTasks >= tableMaxNumTasks) {
- break;
- }
-
- // Skip consuming segments.
- if (tableConfig.getTableType() == TableType.REALTIME && !segmentZKMetadata.getStatus().isCompleted()) {
- continue;
- }
-
- // Skip segments for which a task is already running.
- if (runningSegments.contains(new Segment(tableNameWithType, segmentZKMetadata.getSegmentName()))) {
- continue;
- }
-
- String segmentName = segmentZKMetadata.getSegmentName();
-
- // Skip if the segment is already up-to-date and doesn't have to be refreshed.
- if (!shouldRefreshSegment(segmentZKMetadata, tableConfig, tableStat, schemaStat)) {
- continue;
- }
-
- Map<String, String> configs = new HashMap<>(getBaseTaskConfigs(tableConfig, List.of(segmentName)));
- configs.put(MinionConstants.DOWNLOAD_URL_KEY, segmentZKMetadata.getDownloadUrl());
- configs.put(MinionConstants.UPLOAD_URL_KEY, _clusterInfoAccessor.getVipUrl() + "/segments");
- configs.put(MinionConstants.ORIGINAL_SEGMENT_CRC_KEY, String.valueOf(segmentZKMetadata.getCrc()));
- pinotTaskConfigs.add(new PinotTaskConfig(taskType, configs));
- tableNumTasks++;
+ // Skip consuming segments.
+ if (tableConfig.getTableType() == TableType.REALTIME && !segmentZKMetadata.getStatus().isCompleted()) {
+ continue;
+ }
+
+ // Skip segments for which a task is already running.
+ if (runningSegments.contains(new Segment(tableNameWithType, segmentZKMetadata.getSegmentName()))) {
+ continue;
+ }
+
+ String segmentName = segmentZKMetadata.getSegmentName();
+
+ // Skip if the segment is already up-to-date and doesn't have to be refreshed.
+ if (!shouldRefreshSegment(segmentZKMetadata, tableConfig, tableStat, schemaStat)) {
+ continue;
}
- LOGGER.info("Finished generating {} tasks configs for table: {} " + "for task: {}", tableNumTasks,
- tableNameWithType, taskType);
+ Map<String, String> configs = new HashMap<>(getBaseTaskConfigs(tableConfig, List.of(segmentName)));
+ configs.put(MinionConstants.DOWNLOAD_URL_KEY, segmentZKMetadata.getDownloadUrl());
+ configs.put(MinionConstants.UPLOAD_URL_KEY, _clusterInfoAccessor.getVipUrl() + "/segments");
+ configs.put(MinionConstants.ORIGINAL_SEGMENT_CRC_KEY, String.valueOf(segmentZKMetadata.getCrc()));
+ pinotTaskConfigs.add(new PinotTaskConfig(taskType, configs));
+ tableNumTasks++;
}
+ LOGGER.info("Finished generating {} tasks configs for table: {} " + "for task: {}", tableNumTasks,
+ tableNameWithType, taskType);
Review Comment:
nit: we don't need `+` in the string.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] RefreshSegmentTask - Support for scheduling adhoc tasks [pinot]
Posted by "tibrewalpratik17 (via GitHub)" <gi...@apache.org>.
tibrewalpratik17 commented on code in PR #14596:
URL: https://github.com/apache/pinot/pull/14596#discussion_r1870030763
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/refreshsegment/RefreshSegmentTaskGenerator.java:
##########
@@ -54,84 +54,99 @@ public String getTaskType() {
@Override
public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
- String taskType = RefreshSegmentTask.TASK_TYPE;
List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
- PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
-
- int tableNumTasks = 0;
-
for (TableConfig tableConfig : tableConfigs) {
- String tableNameWithType = tableConfig.getTableName();
- LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
-
// Get the task configs for the table. This is used to restrict the maximum number of allowed tasks per table at
// any given point.
Map<String, String> taskConfigs;
TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
if (tableTaskConfig == null) {
- LOGGER.warn("Failed to find task config for table: {}", tableNameWithType);
+ LOGGER.warn("Failed to find task config for table: {}", tableConfig.getTableName());
continue;
}
taskConfigs = tableTaskConfig.getConfigsForTaskType(RefreshSegmentTask.TASK_TYPE);
- Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableNameWithType);
- int tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- String tableMaxNumTasksConfig = taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
- if (tableMaxNumTasksConfig != null) {
- try {
- tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
- } catch (Exception e) {
- tableMaxNumTasks = RefreshSegmentTask.MAX_NUM_TASKS_PER_TABLE;
- LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and task {}", tableNameWithType, taskType);
- }
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s",
+ tableConfig.getTableName());
+
+ pinotTaskConfigs.addAll(generateTasksForTable(tableConfig, taskConfigs));
+ }
+
+ return pinotTaskConfigs;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(TableConfig tableConfig, Map<String, String> taskConfigs)
+ throws Exception {
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: %s", tableConfig.getTableName());
+ return generateTasksForTable(tableConfig, taskConfigs);
+ }
+
+ private List<PinotTaskConfig> generateTasksForTable(TableConfig tableConfig, Map<String, String> taskConfigs) {
+ String taskType = RefreshSegmentTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+ PinotHelixResourceManager pinotHelixResourceManager = _clusterInfoAccessor.getPinotHelixResourceManager();
+
+ String tableNameWithType = tableConfig.getTableName();
+ LOGGER.info("Start generating RefreshSegment tasks for table: {}", tableNameWithType);
+
Review Comment:
nit: can remove extra line
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org