|
@@ -30,7 +30,6 @@ import java.util.Calendar;
|
|
|
import java.util.List;
|
|
|
import java.util.TimeZone;
|
|
|
|
|
|
-import org.apache.commons.lang3.NotImplementedException;
|
|
|
import org.apache.commons.lang3.StringUtils;
|
|
|
import org.apache.hadoop.conf.Configuration;
|
|
|
import org.apache.hadoop.security.token.delegation.DelegationKey;
|
|
@@ -38,6 +37,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationId;
|
|
|
import org.apache.hadoop.yarn.api.records.ReservationId;
|
|
|
import org.apache.hadoop.yarn.conf.YarnConfiguration;
|
|
|
import org.apache.hadoop.yarn.exceptions.YarnException;
|
|
|
+import org.apache.hadoop.yarn.proto.YarnServerCommonProtos;
|
|
|
import org.apache.hadoop.yarn.security.client.YARNDelegationTokenIdentifier;
|
|
|
import org.apache.hadoop.yarn.server.federation.store.FederationStateStore;
|
|
|
import org.apache.hadoop.yarn.server.federation.store.exception.FederationStateStoreInvalidInputException;
|
|
@@ -102,6 +102,7 @@ import org.apache.hadoop.yarn.server.federation.store.sql.RouterMasterKeyHandler
|
|
|
import org.apache.hadoop.yarn.server.federation.store.sql.RouterStoreTokenHandler;
|
|
|
import org.apache.hadoop.yarn.server.federation.store.sql.RowCountHandler;
|
|
|
import org.apache.hadoop.yarn.server.records.Version;
|
|
|
+import org.apache.hadoop.yarn.server.records.impl.pb.VersionPBImpl;
|
|
|
import org.apache.hadoop.yarn.util.Clock;
|
|
|
import org.apache.hadoop.yarn.util.MonotonicClock;
|
|
|
import org.slf4j.Logger;
|
|
@@ -202,6 +203,12 @@ public class SQLFederationStateStore implements FederationStateStore {
|
|
|
protected static final String CALL_SP_DELETE_DELEGATIONTOKEN =
|
|
|
"{call sp_deleteDelegationToken(?, ?)}";
|
|
|
|
|
|
+ private static final String CALL_SP_STORE_VERSION =
|
|
|
+ "{call sp_storeVersion(?, ?, ?)}";
|
|
|
+
|
|
|
+ private static final String CALL_SP_LOAD_VERSION =
|
|
|
+ "{call sp_getVersion(?, ?)}";
|
|
|
+
|
|
|
private Calendar utcCalendar =
|
|
|
Calendar.getInstance(TimeZone.getTimeZone("UTC"));
|
|
|
|
|
@@ -218,6 +225,8 @@ public class SQLFederationStateStore implements FederationStateStore {
|
|
|
private Connection conn = null;
|
|
|
private int maxAppsInStateStore;
|
|
|
|
|
|
+ protected static final Version CURRENT_VERSION_INFO = Version.newInstance(1, 1);
|
|
|
+
|
|
|
@Override
|
|
|
public void init(Configuration conf) throws YarnException {
|
|
|
driverClass =
|
|
@@ -993,22 +1002,107 @@ public class SQLFederationStateStore implements FederationStateStore {
|
|
|
|
|
|
@Override
|
|
|
public Version getCurrentVersion() {
|
|
|
- throw new NotImplementedException("Code is not implemented");
|
|
|
+ return CURRENT_VERSION_INFO;
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
public Version loadVersion() throws Exception {
|
|
|
- throw new NotImplementedException("Code is not implemented");
|
|
|
+ return getVersion();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Query the Version information of Federation from the database.
|
|
|
+ *
|
|
|
+ * @return Version Info.
|
|
|
+ * @throws Exception Exception Information.
|
|
|
+ */
|
|
|
+ public Version getVersion() throws Exception {
|
|
|
+ CallableStatement callableStatement = null;
|
|
|
+ Version version = null;
|
|
|
+ try {
|
|
|
+ callableStatement = getCallableStatement(CALL_SP_LOAD_VERSION);
|
|
|
+
|
|
|
+ // Set the parameters for the stored procedure
|
|
|
+ callableStatement.registerOutParameter("fedVersion_OUT", java.sql.Types.VARBINARY);
|
|
|
+ callableStatement.registerOutParameter("versionComment_OUT", VARCHAR);
|
|
|
+
|
|
|
+ // Execute the query
|
|
|
+ long startTime = clock.getTime();
|
|
|
+ callableStatement.executeUpdate();
|
|
|
+ long stopTime = clock.getTime();
|
|
|
+
|
|
|
+ // Parsing version information.
|
|
|
+ String versionComment = callableStatement.getString("versionComment_OUT");
|
|
|
+ byte[] fedVersion = callableStatement.getBytes("fedVersion_OUT");
|
|
|
+ if (versionComment != null && fedVersion != null) {
|
|
|
+ version = new VersionPBImpl(YarnServerCommonProtos.VersionProto.parseFrom(fedVersion));
|
|
|
+ FederationStateStoreClientMetrics.succeededStateStoreCall(stopTime - startTime);
|
|
|
+ }
|
|
|
+ } catch (SQLException e) {
|
|
|
+ FederationStateStoreClientMetrics.failedStateStoreCall();
|
|
|
+ FederationStateStoreUtils.logAndThrowRetriableException(e, LOG,
|
|
|
+ "Unable to select the version.");
|
|
|
+ } finally {
|
|
|
+ // Return to the pool the CallableStatement
|
|
|
+ FederationStateStoreUtils.returnToPool(LOG, callableStatement);
|
|
|
+ }
|
|
|
+ return version;
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
public void storeVersion() throws Exception {
|
|
|
- throw new NotImplementedException("Code is not implemented");
|
|
|
+ byte[] fedVersion = ((VersionPBImpl) CURRENT_VERSION_INFO).getProto().toByteArray();
|
|
|
+ String versionComment = CURRENT_VERSION_INFO.toString();
|
|
|
+ storeVersion(fedVersion, versionComment);
|
|
|
}
|
|
|
|
|
|
- @Override
|
|
|
- public void checkVersion() throws Exception {
|
|
|
- throw new NotImplementedException("Code is not implemented");
|
|
|
+ /**
|
|
|
+ * Store the Federation Version in the database.
|
|
|
+ *
|
|
|
+ * @param fedVersion Federation Version.
|
|
|
+ * @param versionComment Federation Version Comment,
|
|
|
+ * We use the result of Version toString as version Comment.
|
|
|
+ * @throws YarnException indicates exceptions from yarn servers.
|
|
|
+ */
|
|
|
+ public void storeVersion(byte[] fedVersion, String versionComment) throws YarnException {
|
|
|
+ CallableStatement callableStatement = null;
|
|
|
+
|
|
|
+ try {
|
|
|
+ callableStatement = getCallableStatement(CALL_SP_STORE_VERSION);
|
|
|
+
|
|
|
+ // Set the parameters for the stored procedure
|
|
|
+ callableStatement.setBytes("fedVersion_IN", fedVersion);
|
|
|
+ callableStatement.setString("versionComment_IN", versionComment);
|
|
|
+ callableStatement.registerOutParameter("rowCount_OUT", INTEGER);
|
|
|
+
|
|
|
+ // Execute the query
|
|
|
+ long startTime = clock.getTime();
|
|
|
+ callableStatement.executeUpdate();
|
|
|
+ long stopTime = clock.getTime();
|
|
|
+
|
|
|
+ // Check the ROWCOUNT value, if it is equal to 0 it means the call
|
|
|
+ // did not add a new version into FederationStateStore
|
|
|
+ int rowCount = callableStatement.getInt("rowCount_OUT");
|
|
|
+ if (rowCount == 0) {
|
|
|
+ FederationStateStoreUtils.logAndThrowStoreException(LOG,
|
|
|
+ "The version %s was not insert into the StateStore.", versionComment);
|
|
|
+ }
|
|
|
+ // Check the ROWCOUNT value, if it is different from 1 it means the call
|
|
|
+ // had a wrong behavior. Maybe the database is not set correctly.
|
|
|
+ if (rowCount != 1) {
|
|
|
+ FederationStateStoreUtils.logAndThrowStoreException(LOG,
|
|
|
+ "Wrong behavior during insert the version %s.", versionComment);
|
|
|
+ }
|
|
|
+ FederationStateStoreClientMetrics.succeededStateStoreCall(stopTime - startTime);
|
|
|
+ LOG.info("Insert into the state store the version : {}.", versionComment);
|
|
|
+ } catch (SQLException e) {
|
|
|
+ FederationStateStoreClientMetrics.failedStateStoreCall();
|
|
|
+ FederationStateStoreUtils.logAndThrowRetriableException(e, LOG,
|
|
|
+ "Unable to insert the newly version : %s.", versionComment);
|
|
|
+ } finally {
|
|
|
+ // Return to the pool the CallableStatement
|
|
|
+ FederationStateStoreUtils.returnToPool(LOG, callableStatement);
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
@Override
|