This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 38c73aa Fix the issue of scheduling task on all tables (#4563)
38c73aa is described below
commit 38c73aa1c8d146fc1f344cb90a10a884b8734d6b
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Sat Aug 24 21:22:37 2019 -0700
Fix the issue of scheduling task on all tables (#4563)
ControllerPeriodicTask.processTables() should only process the tables that
are managed by the controller
PinotTaskManager is overriding processTables() to schedule tasks
---
.../controller/helix/core/minion/PinotTaskManager.java | 11 +++++------
.../helix/core/periodictask/ControllerPeriodicTask.java | 14 +++++++++-----
2 files changed, 14 insertions(+), 11 deletions(-)
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java
index 649b966..8ceffff 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java
@@ -90,11 +90,10 @@ public class PinotTaskManager extends
ControllerPeriodicTask<Void> {
public synchronized Map<String, String> scheduleTasks() {
Map<String, String> tasksScheduled =
scheduleTasks(_pinotHelixResourceManager.getAllTables());
- // NOTE: this method might be called from the Rest API instead of the
periodic task scheduler on non-leader
- // controllers, so if the task is stopped (non-leader controller), clean
up the task
- if (!isStarted()) {
- cleanUpTask();
- }
+ // Reset the task because this method will be called from the Rest API
instead of the periodic task scheduler
+ // TODO: Clean up only the non-leader tables instead of all tables
+ cleanUpTask();
+ setUpTask();
return tasksScheduled;
}
@@ -105,7 +104,7 @@ public class PinotTaskManager extends
ControllerPeriodicTask<Void> {
* @param tableNamesWithType List of table names with type suffix
* @return Map from task type to task scheduled
*/
- private Map<String, String> scheduleTasks(List<String> tableNamesWithType) {
+ private synchronized Map<String, String> scheduleTasks(List<String>
tableNamesWithType) {
_controllerMetrics.addMeteredGlobalValue(ControllerMeter.NUMBER_TIMES_SCHEDULE_TASKS_CALLED,
1L);
Set<String> taskTypes = _taskGeneratorRegistry.getAllTaskTypes();
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTask.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTask.java
index a782385..d2c2b13 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTask.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTask.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.controller.helix.core.periodictask;
+import java.util.ArrayList;
import java.util.List;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.pinot.common.metrics.ControllerGauge;
@@ -57,7 +58,14 @@ public abstract class ControllerPeriodicTask<C> extends
BasePeriodicTask {
protected final void runTask() {
_controllerMetrics.addMeteredTableValue(_taskName,
ControllerMeter.CONTROLLER_PERIODIC_TASK_RUN, 1L);
try {
- processTables(_pinotHelixResourceManager.getAllTables());
+ // Process the tables that are managed by this controller
+ List<String> tablesToProcess = new ArrayList<>();
+ for (String tableNameWithType :
_pinotHelixResourceManager.getAllTables()) {
+ if (_leadControllerManager.isLeaderForTable(tableNameWithType)) {
+ tablesToProcess.add(tableNameWithType);
+ }
+ }
+ processTables(tablesToProcess);
} catch (Exception e) {
LOGGER.error("Caught exception while running task: {}", _taskName, e);
_controllerMetrics.addMeteredTableValue(_taskName,
ControllerMeter.CONTROLLER_PERIODIC_TASK_ERROR, 1L);
@@ -79,10 +87,6 @@ public abstract class ControllerPeriodicTask<C> extends
BasePeriodicTask {
LOGGER.info("Task: {} is stopped, early terminate the task",
_taskName);
break;
}
- // Check if current controller is the leader for this table.
- if (!_leadControllerManager.isLeaderForTable(tableNameWithType)) {
- continue;
- }
try {
processTable(tableNameWithType, context);
} catch (Exception e) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
