You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "Will-Lo (via GitHub)" <gi...@apache.org> on 2023/05/31 15:53:53 UTC
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3700: [GOBBLIN-1837] Implement multi-active, non blocking for leader host
Will-Lo commented on code in PR #3700:
URL: https://github.com/apache/gobblin/pull/3700#discussion_r1211927683
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MysqlSchedulerLeaseDeterminationStore.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.runtime.api;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+
+import com.google.inject.Inject;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.util.ConfigUtils;
+
+
+public class MysqlSchedulerLeaseDeterminationStore implements SchedulerLeaseDeterminationStore {
Review Comment:
Can you add a javadoc describing what this class does?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java:
##########
@@ -151,7 +170,16 @@ protected void processMessage(DecodeableKafkaRecord message) {
log.info("Received insert dag action and about to send kill flow request");
dagManager.handleKillFlowRequest(flowGroup, flowName, Long.parseLong(flowExecutionId));
this.killsInvoked.mark();
- } else {
+ } else if (dagAction.equals(DagActionStore.DagActionValue.LAUNCH)) {
+ // If multi-active scheduler is NOT turned on we should not receive these type of events
+ if (!this.isMultiActiveSchedulerEnabled) {
+ log.warn("Received LAUNCH dagAction while not in multi-active scheduler mode for flow group: {}, flow name:"
+ + "{}, execution id: {}, dagAction: {}", flowGroup, flowName, flowExecutionId, dagAction);
+ this.unexpectedErrors.mark();
+ }
+ log.info("Received insert dag action and about to forward launch request to DagManager");
+ submitFlowToDagManager(flowGroup, flowName);
+ }else {
Review Comment:
nit: missing space
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/DagActionStore.java:
##########
@@ -27,7 +27,9 @@
public interface DagActionStore {
enum DagActionValue {
KILL,
- RESUME
+ RESUME,
+ // TODO: potentially combine this enum with {@link SchedulerLeaseDeterminationStore.FlowActionType}
+ LAUNCH
Review Comment:
Naive question for the sake of my understanding here:
So we are making the assumption that the host scheduler that wins the lease to run the job does not imply that the host actually runs the job here? So any host can run a job that is stored/scheduled by other hosts.
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MysqlSchedulerLeaseDeterminationStore.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.runtime.api;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+
+import com.google.inject.Inject;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.util.ConfigUtils;
+
+
+public class MysqlSchedulerLeaseDeterminationStore implements SchedulerLeaseDeterminationStore {
+ public static final String CONFIG_PREFIX = "MysqlSchedulerLeaseDeterminationStore";
+
+ protected final DataSource dataSource;
+ private final DagActionStore dagActionStore;
+ private final String tableName;
+ private final long epsilon;
+ private final long linger;
+ /* TODO:
+ - define retention on this table
+ - initialize table with epsilon and linger if one already doesn't exist using these configs
+ - join with table above to ensure epsilon/linger values are consistent across hosts (in case hosts are deployed with different configs)
+ */
+ protected static final String WHERE_CLAUSE_TO_MATCH_ROW = "WHERE flow_group=? AND flow_name=? AND flow_execution_id=? "
+ + "AND flow_action=? AND ABS(trigger_event_timestamp-?) <= %s";
+ protected static final String ATTEMPT_INSERT_AND_GET_PURSUANT_TIMESTAMP_STATEMENT = "INSERT INTO %s (flow_group, "
+ + "flow_name, flow_execution_id, flow_action, trigger_event_timestamp) VALUES (?, ?, ?, ?, ?) WHERE NOT EXISTS ("
+ + "SELECT * FROM %s " + WHERE_CLAUSE_TO_MATCH_ROW + "; SELECT ROW_COUNT() AS rows_inserted_count, "
+ + "pursuant_timestamp FROM %s " + WHERE_CLAUSE_TO_MATCH_ROW;
+
+ protected static final String UPDATE_PURSUANT_TIMESTAMP_STATEMENT = "UPDATE %s SET pursuant_timestamp = NULL "
+ + WHERE_CLAUSE_TO_MATCH_ROW;
+ private static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %S (" + "flow_group varchar("
+ + ServiceConfigKeys.MAX_FLOW_GROUP_LENGTH + ") NOT NULL, flow_name varchar("
+ + ServiceConfigKeys.MAX_FLOW_GROUP_LENGTH + ") NOT NULL, " + "flow_execution_id varchar("
+ + ServiceConfigKeys.MAX_FLOW_EXECUTION_ID_LENGTH + ") NOT NULL, flow_action varchar(100) NOT NULL, "
+ + "trigger_event_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, "
+ + "pursuant_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,"
+ + "PRIMARY KEY (flow_group,flow_name,flow_execution_id,flow_action,trigger_event_timestamp)";
+
+ @Inject
+ public MysqlSchedulerLeaseDeterminationStore(Config config, DagActionStore dagActionStore) throws IOException {
+ if (config.hasPath(CONFIG_PREFIX)) {
+ config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+ } else {
+ throw new IOException("Please specify the config for MysqlSchedulerLeaseDeterminationStore");
+ }
+
+ this.tableName = ConfigUtils.getString(config, ConfigurationKeys.SCHEDULER_LEASE_DETERMINATION_STORE_DB_TABLE_KEY,
+ ConfigurationKeys.DEFAULT_SCHEDULER_LEASE_DETERMINATION_STORE_DB_TABLE);
+ this.epsilon = ConfigUtils.getLong(config, ConfigurationKeys.SCHEDULER_TRIGGER_EVENT_EPSILON_MILLIS_KEY,
+ ConfigurationKeys.DEFAULT_SCHEDULER_TRIGGER_EVENT_EPSILON_MILLIS);
+ this.linger = ConfigUtils.getLong(config, ConfigurationKeys.SCHEDULER_TRIGGER_EVENT_EPSILON_MILLIS_KEY,
+ ConfigurationKeys.DEFAULT_SCHEDULER_TRIGGER_EVENT_EPSILON_MILLIS);
+
+ this.dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+ try (Connection connection = dataSource.getConnection();
+ PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+ createStatement.executeUpdate();
+ connection.commit();
+ } catch (SQLException e) {
+ throw new IOException("Table creation failure for " + tableName, e);
+ }
+ this.dagActionStore = dagActionStore;
+ }
+
+ @Override
+ public LeaseAttemptStatus attemptInsertAndGetPursuantTimestamp(String flowGroup, String flowName,
+ String flowExecutionId, FlowActionType flowActionType, long triggerTimeMillis)
+ throws IOException {
+ Timestamp triggerTimestamp = new Timestamp(triggerTimeMillis);
+ try (Connection connection = this.dataSource.getConnection();
+ PreparedStatement insertStatement = connection.prepareStatement(
+ String.format(ATTEMPT_INSERT_AND_GET_PURSUANT_TIMESTAMP_STATEMENT, tableName, tableName, epsilon, tableName,
+ epsilon))) {
+ int i = 0;
+ // Values to set in new row
+ insertStatement.setString(++i, flowGroup);
+ insertStatement.setString(++i, flowName);
+ insertStatement.setString(++i, flowExecutionId);
+ insertStatement.setString(++i, flowActionType.toString());
+ insertStatement.setTimestamp(++i, triggerTimestamp);
+ // Values to check if existing row matches
+ insertStatement.setString(++i, flowGroup);
+ insertStatement.setString(++i, flowName);
+ insertStatement.setString(++i, flowExecutionId);
+ insertStatement.setString(++i, flowActionType.toString());
+ insertStatement.setTimestamp(++i, triggerTimestamp);
+ // Values to make select statement to read row
+ insertStatement.setString(++i, flowGroup);
+ insertStatement.setString(++i, flowName);
+ insertStatement.setString(++i, flowExecutionId);
+ insertStatement.setString(++i, flowActionType.toString());
+ insertStatement.setTimestamp(++i, triggerTimestamp);
+ ResultSet resultSet = insertStatement.executeQuery();
+ connection.commit();
+
+ if (!resultSet.next()) {
+ resultSet.close();
+ throw new IOException(String.format("Unexpected error where no result returned while trying to obtain lease. "
+ + "This error indicates that no entry existed for trigger flow event for table %s flow group: %s, flow "
+ + "name: %s flow execution id: %s and trigger timestamp: %s when one should have been inserted",
+ tableName, flowGroup, flowName, flowExecutionId, triggerTimestamp));
+ }
+ // If a row was inserted, then we have obtained the lease
+ int rowsUpdated = resultSet.getInt(1);
+ if (rowsUpdated == 1) {
+ // If the pursuing flow launch has been persisted to the {@link DagActionStore} we have completed lease obtainment
+ this.dagActionStore.addDagAction(flowGroup, flowName, flowExecutionId, DagActionStore.DagActionValue.LAUNCH);
+ if (this.dagActionStore.exists(flowGroup, flowName, flowExecutionId, DagActionStore.DagActionValue.LAUNCH)) {
+ if (updatePursuantTimestamp(flowGroup, flowName, flowExecutionId, flowActionType, triggerTimestamp)) {
+ // TODO: potentially add metric here to count number of flows scheduled by each scheduler
Review Comment:
I think this would be an excellent metric also for the purpose of loadbalancing, maybe worth implementing directly
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/SchedulerLeaseAlgoHandler.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
+import java.util.Locale;
+import java.util.Properties;
+import java.util.Random;
+
+import org.quartz.JobKey;
+import org.quartz.SchedulerException;
+import org.quartz.Trigger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.typesafe.config.Config;
+
+import javax.inject.Inject;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.SchedulerLeaseDeterminationStore;
+import org.apache.gobblin.scheduler.JobScheduler;
+import org.apache.gobblin.scheduler.SchedulerService;
+import org.apache.gobblin.util.ConfigUtils;
+
+
+public class SchedulerLeaseAlgoHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(SchedulerLeaseAlgoHandler.class);
+ private final long linger;
+ private final int staggerUpperBoundSec;
+ private static Random random = new Random();
+ protected SchedulerLeaseDeterminationStore leaseDeterminationStore;
+ protected JobScheduler jobScheduler;
+ protected SchedulerService schedulerService;
+ @Inject
+ public SchedulerLeaseAlgoHandler(Config config, SchedulerLeaseDeterminationStore leaseDeterminationStore,
+ JobScheduler jobScheduler, SchedulerService schedulerService)
+ throws IOException {
+ this.linger = ConfigUtils.getLong(config, ConfigurationKeys.SCHEDULER_TRIGGER_EVENT_EPSILON_MILLIS_KEY,
+ ConfigurationKeys.DEFAULT_SCHEDULER_TRIGGER_EVENT_EPSILON_MILLIS);
+ this.staggerUpperBoundSec = ConfigUtils.getInt(config,
+ ConfigurationKeys.SCHEDULER_STAGGERING_UPPER_BOUND_SEC_KEY,
+ ConfigurationKeys.DEFAULT_SCHEDULER_STAGGERING_UPPER_BOUND_SEC);
+ this.leaseDeterminationStore = leaseDeterminationStore;
+ this.jobScheduler = jobScheduler;
+ this.schedulerService = schedulerService;
+ }
+ private SchedulerLeaseDeterminationStore schedulerLeaseDeterminationStore;
+
+ /**
+ * This method is used in the multi-active scheduler case for one or more hosts to respond to a flow's trigger event
+ * by attempting a lease for the flow event.
+ * @param jobProps
+ * @param flowGroup
+ * @param flowName
+ * @param flowExecutionId
+ * @param flowActionType
+ * @param triggerTimeMillis
+ * @return true if this host obtained the lease for this flow's trigger event, false otherwise.
+ * @throws IOException
+ */
+ public boolean handleNewTriggerEvent(Properties jobProps, String flowGroup, String flowName, String flowExecutionId,
+ SchedulerLeaseDeterminationStore.FlowActionType flowActionType, long triggerTimeMillis)
+ throws IOException {
+ SchedulerLeaseDeterminationStore.LeaseAttemptStatus leaseAttemptStatus =
+ schedulerLeaseDeterminationStore.attemptInsertAndGetPursuantTimestamp(flowGroup, flowName, flowExecutionId,
+ flowActionType, triggerTimeMillis);
+ // TODO: add a log event or metric for each of these cases
+ switch (leaseAttemptStatus) {
+ case LEASE_OBTAINED:
+ return true;
+ case PREVIOUS_LEASE_EXPIRED:
+ // recursively try obtaining lease again immediately, stops when reaches one of the other cases
+ return handleNewTriggerEvent(jobProps, flowGroup, flowName, flowExecutionId, flowActionType, triggerTimeMillis);
+ case PREVIOUS_LEASE_VALID:
+ scheduleReminderForTriggerEvent(jobProps, flowGroup, flowName, flowExecutionId, flowActionType, triggerTimeMillis);
+ }
+ return false;
+ }
+
+ /**
+ * This method is used by {@link SchedulerLeaseAlgoHandler.handleNewTriggerEvent} to schedule a reminder for itself to
+ * check on the other participant's progress during pursuing orchestration after the time the lease should expire.
+ * If the previous participant was successful, then no further action is taken otherwise we re-attempt pursuing
+ * orchestration ourselves.
+ * @param flowGroup
+ * @param flowName
+ * @param flowExecutionId
+ * @param flowActionType
+ * @param triggerTimeMillis
+ */
+ protected void scheduleReminderForTriggerEvent(Properties jobProps, String flowGroup, String flowName, String flowExecutionId,
+ SchedulerLeaseDeterminationStore.FlowActionType flowActionType, long triggerTimeMillis) {
+ // Check-in `linger` time after the current timestamp which is "close-enough" to the time the pursuant attempted
+ // the flow action. We also add a small randomization to avoid 'thundering herd' issue
+ String cronExpression = createCronFromDelayPeriod(linger + random.nextInt(staggerUpperBoundSec));
+ jobProps.setProperty(ConfigurationKeys.JOB_SCHEDULE_KEY, cronExpression);
+ // This timestamp is what will be used to identify the particular flow trigger event it's associated with
+ jobProps.setProperty(ConfigurationKeys.SCHEDULER_ORIGINAL_TRIGGER_TIMESTAMP_MILLIS_KEY, String.valueOf(triggerTimeMillis));
+ JobKey key = new JobKey(flowName, flowGroup);
+ Trigger trigger = this.jobScheduler.getTrigger(key, jobProps);
+ try {
+ LOG.info("Attempting to add job reminder to Scheduler Service where job is %s trigger event %s and reminder is at "
+ + "%s.", key, triggerTimeMillis, trigger.getNextFireTime());
+ this.schedulerService.getScheduler().scheduleJob(trigger);
+ } catch (SchedulerException e) {
+ LOG.warn("Failed to add job reminder due to SchedulerException for job %s trigger event %s ", key, triggerTimeMillis, e);
+ }
+ LOG.info(String.format("Scheduled reminder for job %s trigger event %s. Next run: %s.", key, triggerTimeMillis, trigger.getNextFireTime()));
+ }
+
+ /**
+ * These methods should only be called from the Orchestrator or JobScheduler classes as it directly adds jobs to the
+ * Quartz scheduler
+ * @param delayPeriodSeconds
+ * @return
+ */
+ protected static String createCronFromDelayPeriod(long delayPeriodSeconds) {
+ LocalDateTime now = LocalDateTime.now();
+ LocalDateTime delaySecondsLater = now.plus(delayPeriodSeconds, ChronoUnit.SECONDS);
+ // TODO: investigate potentially better way of generating cron expression that does not make it US dependent
+ DateTimeFormatter formatter = DateTimeFormatter.ofPattern("ss mm HH dd MM ? yyyy", Locale.US);
Review Comment:
Does LocalDateTime.now() default to US timezone? Otherwise this could cause issues. I would suggest you use LocalDateTime.now(<timezone>) to ensure consistency in this system across timezones.
Also, I think GaaS scheduler defaults to UTC.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java:
##########
@@ -151,7 +170,16 @@ protected void processMessage(DecodeableKafkaRecord message) {
log.info("Received insert dag action and about to send kill flow request");
dagManager.handleKillFlowRequest(flowGroup, flowName, Long.parseLong(flowExecutionId));
this.killsInvoked.mark();
- } else {
+ } else if (dagAction.equals(DagActionStore.DagActionValue.LAUNCH)) {
+ // If multi-active scheduler is NOT turned on we should not receive these type of events
+ if (!this.isMultiActiveSchedulerEnabled) {
+ log.warn("Received LAUNCH dagAction while not in multi-active scheduler mode for flow group: {}, flow name:"
+ + "{}, execution id: {}, dagAction: {}", flowGroup, flowName, flowExecutionId, dagAction);
+ this.unexpectedErrors.mark();
+ }
+ log.info("Received insert dag action and about to forward launch request to DagManager");
+ submitFlowToDagManager(flowGroup, flowName);
Review Comment:
I agree with this sentiment, rather would let it fail loudly if it's not expected.
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org