diff --git a/extensions/mssql/src/objectExplorerNodeProvider/objectExplorerNodeProvider.ts b/extensions/mssql/src/objectExplorerNodeProvider/objectExplorerNodeProvider.ts
index dd925b99ad..cde5c97ba3 100644
--- a/extensions/mssql/src/objectExplorerNodeProvider/objectExplorerNodeProvider.ts
+++ b/extensions/mssql/src/objectExplorerNodeProvider/objectExplorerNodeProvider.ts
@@ -255,7 +255,7 @@ export class SqlClusterSession {
 
 	public async getSqlClusterConnection(): Promise<SqlClusterConnection> {
 		if (!this._sqlClusterConnection) {
-			const sqlClusterConnectionParams = await getSqlClusterConnectionParams(this._sqlConnectionProfile);
+			const sqlClusterConnectionParams = await getSqlClusterConnectionParams(this._sqlConnectionProfile, this._appContext);
 			this._sqlClusterConnection = new SqlClusterConnection(sqlClusterConnectionParams);
 		}
 		return this._sqlClusterConnection;
diff --git a/extensions/mssql/src/sparkFeature/dialog/dialogCommands.ts b/extensions/mssql/src/sparkFeature/dialog/dialogCommands.ts
index 2bb03c91ca..861171644a 100644
--- a/extensions/mssql/src/sparkFeature/dialog/dialogCommands.ts
+++ b/extensions/mssql/src/sparkFeature/dialog/dialogCommands.ts
@@ -103,7 +103,7 @@ export class OpenSparkJobSubmissionDialogCommand extends Command {
 		let sqlConnection = connectionMap.get(selectedHost);
 		if (!sqlConnection) { throw new Error(errorMsg); }
 
-		let sqlClusterConnection = await SqlClusterLookUp.getSqlClusterConnectionParams(sqlConnection);
+		let sqlClusterConnection = await SqlClusterLookUp.getSqlClusterConnectionParams(sqlConnection, this.appContext);
 		if (!sqlClusterConnection) {
 			throw new Error(localize('errorNotSqlBigDataCluster', "The selected server does not belong to a SQL Server Big Data Cluster"));
 		}
diff --git a/extensions/mssql/src/sqlClusterLookUp.ts b/extensions/mssql/src/sqlClusterLookUp.ts
index 3ac8c963d3..30867e888b 100644
--- a/extensions/mssql/src/sqlClusterLookUp.ts
+++ b/extensions/mssql/src/sqlClusterLookUp.ts
@@ -53,7 +53,8 @@ async function findSqlClusterConnectionBySqlConnProfile(sqlConnProfile: azdata.I
 }
 
 export async function getSqlClusterConnectionParams(
-	obj: azdata.IConnectionProfile | azdata.connection.Connection | ICommandObjectExplorerContext): Promise<ConnectionParam> {
+	obj: azdata.IConnectionProfile | azdata.connection.Connection | ICommandObjectExplorerContext,
+	appContext: AppContext): Promise<ConnectionParam> {
 
 	if (!obj) { return undefined; }
 
@@ -62,16 +63,16 @@ export async function getSqlClusterConnectionParams(
 		if (obj.providerName === constants.mssqlClusterProviderName) {
 			sqlClusterConnInfo = 'id' in obj ? connProfileToConnectionParam(obj) : connToConnectionParam(obj);
 		} else {
-			sqlClusterConnInfo = await createSqlClusterConnInfo(obj);
+			sqlClusterConnInfo = await createSqlClusterConnInfo(obj, appContext);
 		}
 	} else {
-		sqlClusterConnInfo = await createSqlClusterConnInfo(obj.explorerContext.connectionProfile);
+		sqlClusterConnInfo = await createSqlClusterConnInfo(obj.explorerContext.connectionProfile, appContext);
 	}
 
 	return sqlClusterConnInfo;
 }
 
-async function createSqlClusterConnInfo(sqlConnInfo: azdata.IConnectionProfile | azdata.connection.Connection): Promise<ConnectionParam> {
+async function createSqlClusterConnInfo(sqlConnInfo: azdata.IConnectionProfile | azdata.connection.Connection, appContext: AppContext): Promise<ConnectionParam> {
 	if (!sqlConnInfo) { return undefined; }
 
 	let connectionId: string = 'id' in sqlConnInfo ? sqlConnInfo.id : sqlConnInfo.connectionId;
@@ -96,11 +97,19 @@ async function createSqlClusterConnInfo(sqlConnInfo: azdata.IConnectionProfile |
 	let authType = clusterConnInfo.options[constants.authenticationTypePropName] = sqlConnInfo.options[constants.authenticationTypePropName];
 	const controllerEndpoint = endpoints.find(ep => ep.name.toLowerCase() === 'controller');
 	if (authType && authType.toLowerCase() !== constants.integratedAuth) {
-		clusterConnInfo.options[constants.userPropName] = sqlConnInfo.options[constants.userPropName]; //should be the same user as sql master
-		clusterConnInfo.options[constants.passwordPropName] = credentials.password;
+		const usernameKey = `bdc.username::${connectionId}`;
+		const savedUsername = appContext.extensionContext.globalState.get(usernameKey);
+		const credentialProvider = await azdata.credentials.getProvider('mssql.bdc.password');
+		const savedPassword = (await credentialProvider.readCredential(connectionId)).password;
+		// If we don't have a previously saved username/password then use the SQL connection credentials as a best guess,
+		// if those don't work then we'll prompt the user for the info
+		clusterConnInfo.options[constants.userPropName] = savedUsername ?? sqlConnInfo.options[constants.userPropName];
+		clusterConnInfo.options[constants.passwordPropName] = savedPassword ?? credentials.password;
 		try {
 			clusterController = await getClusterController(controllerEndpoint.endpoint, clusterConnInfo);
-
+			// We've successfully connected so now store the username/password for future connections
+			appContext.extensionContext.globalState.update(usernameKey, clusterConnInfo.options[constants.userPropName]);
+			credentialProvider.saveCredential(connectionId, clusterConnInfo.options[constants.passwordPropName]);
 			clusterConnInfo.options[constants.userPropName] = await clusterController.getKnoxUsername(clusterConnInfo.options[constants.userPropName]);
 		} catch (err) {
 			console.log(`Unexpected error getting Knox username for SQL Cluster connection: ${err}`);