Polishing.

Introduce configuration option to disable schema metadata during schema actions. Tweak method naming. Apply schema metadata suspension also to destroy methods.

Related ticket: #990
Related ticket: #1253

Original pull request: #1255.

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java

@@ -23,6 +23,7 @@ import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CompletionStage;
 import java.util.function.IntFunction;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -57,6 +58,7 @@ import org.springframework.util.StringUtils;
 
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.CqlSessionBuilder;
+import com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures;
 
 /**
  * Factory for creating and configuring a Cassandra {@link CqlSession}, which is a thread-safe singleton. As such, it is
@@ -108,6 +110,8 @@ public class CqlSessionFactoryBean
 
 	private SchemaAction schemaAction = SchemaAction.NONE;
 
+	private boolean suspendLifecycleSchemaRefresh = false;
+
 	private @Nullable SessionBuilderConfigurer sessionBuilderConfigurer;
 
 	private IntFunction<Collection<InetSocketAddress>> contactPoints = port -> createInetSocketAddresses(
@@ -348,8 +352,8 @@ public class CqlSessionFactoryBean
 	 * Set the {@link SchemaAction}.
 	 *
 	 * @param schemaAction must not be {@literal null}.
-	 * @deprecated Use {@link CassandraSessionFactoryBean} with
-	 *             {@link CassandraSessionFactoryBean#setSchemaAction(SchemaAction)} instead.
+	 * @deprecated Use {@link SessionFactoryFactoryBean} with
+	 *             {@link SessionFactoryFactoryBean#setSchemaAction(SchemaAction)} instead.
 	 */
 	@Deprecated
 	public void setSchemaAction(SchemaAction schemaAction) {
@@ -366,6 +370,24 @@ public class CqlSessionFactoryBean
 		return this.schemaAction;
 	}
 
+	/**
+	 * Set whether to suspend schema refresh settings during {@link #afterPropertiesSet()} and {@link #destroy()}
+	 * lifecycle callbacks. Disabled by default to use schema metadata settings of the session configuration. When enabled
+	 * (set to {@code true}), then schema refresh during lifecycle methods is suspended until finishing schema actions to
+	 * avoid periodic schema refreshes for each DDL statement.
+	 * <p>
+	 * Suspending schema refresh can be useful to delay schema agreement until the entire schema is created. Note that
+	 * disabling schema refresh may interfere with schema actions. {@link SchemaAction#RECREATE_DROP_UNUSED} and
+	 * mapping-based schema creation rely on schema metadata.
+	 *
+	 * @param suspendLifecycleSchemaRefresh {@code true} to suspend the schema refresh during lifecycle callbacks;
+	 *          {@code false} otherwise to retain the session schema refresh configuration.
+	 * @since 2.7
+	 */
+	public void setSuspendLifecycleSchemaRefresh(boolean suspendLifecycleSchemaRefresh) {
+		this.suspendLifecycleSchemaRefresh = suspendLifecycleSchemaRefresh;
+	}
+
 	/**
 	 * Returns a reference to the connected Cassandra {@link CqlSession}.
 	 *
@@ -451,20 +473,30 @@ public class CqlSessionFactoryBean
 
 		this.session = buildSession(sessionBuilder);
 
-		try {
-			SchemaRefreshUtils.withDisabledSchema(this.session, () -> {
-				executeCql(getStartupScripts().stream(), this.session);
-				performSchemaAction();
-			});
-		} catch (RuntimeException e) {
-			throw e;
-		} catch (Exception e) {
-			throw new IllegalStateException("Unexpected checked exception thrown", e);
+		initializeSchema(this.systemSession, this.session);
+	}
+
+	private void initializeSchema(CqlSession systemSession, CqlSession session) {
+
+		Runnable schemaActionRunnable = () -> {
+
+			executeCql(getStartupScripts().stream(), session);
+			performSchemaAction();
+		};
+
+		List<CompletionStage<?>> futures = new ArrayList<>(2);
+
+		if (this.suspendLifecycleSchemaRefresh) {
+			futures.add(SchemaUtils.withSuspendedAsyncSchemaRefresh(session, schemaActionRunnable));
+		} else {
+			futures.add(SchemaUtils.withAsyncSchemaRefresh(session, schemaActionRunnable));
 		}
 
-		if (this.systemSession.isSchemaMetadataEnabled()) {
-			this.systemSession.refreshSchema();
+		if (systemSession.isSchemaMetadataEnabled()) {
+			futures.add(systemSession.refreshSchemaAsync());
 		}
+
+		futures.forEach(CompletableFutures::getUninterruptibly);
 	}
 
 	protected CqlSessionBuilder buildBuilder() {
@@ -532,7 +564,15 @@ public class CqlSessionFactoryBean
 		keyspaceStartupSpecifications.addAll(this.keyspaceCreations);
 		keyspaceStartupSpecifications.addAll(this.keyspaceAlterations);
 
-		executeSpecificationsAndScripts(keyspaceStartupSpecifications, this.keyspaceStartupScripts, session);
+		Runnable schemaActionRunnable = () -> {
+			executeSpecificationsAndScripts(keyspaceStartupSpecifications, this.keyspaceStartupScripts, session);
+		};
+
+		if (this.suspendLifecycleSchemaRefresh) {
+			SchemaUtils.withSuspendedAsyncSchemaRefresh(session, schemaActionRunnable);
+		} else {
+			schemaActionRunnable.run();
+		}
 	}
 
 	/**
@@ -560,7 +600,7 @@ public class CqlSessionFactoryBean
 	}
 
 	/**
-	 * Perform the configure {@link SchemaAction} using {@link CassandraMappingContext} metadata.
+	 * Perform the configured {@link SchemaAction} using {@link CassandraMappingContext} metadata.
 	 */
 	protected void performSchemaAction() {
 
@@ -644,8 +684,23 @@ public class CqlSessionFactoryBean
 	public void destroy() {
 
 		if (this.session != null) {
-			executeCql(getShutdownScripts().stream(), this.session);
-			executeSpecificationsAndScripts(this.keyspaceDrops, this.keyspaceShutdownScripts, this.systemSession);
+
+			Runnable schemaActionRunnable = () -> {
+				executeCql(getShutdownScripts().stream(), this.session);
+			};
+
+			Runnable systemSchemaActionRunnable = () -> {
+				executeSpecificationsAndScripts(this.keyspaceDrops, this.keyspaceShutdownScripts, this.systemSession);
+			};
+
+			if (this.suspendLifecycleSchemaRefresh) {
+				SchemaUtils.withSuspendedAsyncSchemaRefresh(this.session, schemaActionRunnable);
+				SchemaUtils.withSuspendedAsyncSchemaRefresh(this.systemSession, systemSchemaActionRunnable);
+			} else {
+				schemaActionRunnable.run();
+				systemSchemaActionRunnable.run();
+			}
+
 			closeSession();
 			closeSystemSession();
 		}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SchemaRefreshUtils.java

@@ -1,48 +0,0 @@
-/*
- * Copyright 2022 the original author or authors.
- *
- * Licensed 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
- *
- *      https://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.springframework.data.cassandra.config;
-
-import com.datastax.oss.driver.api.core.session.Session;
-
-/**
- * Utility methods for executing schema actions with refresh disabled.
- *
- * @author Ammar Khaku
- */
-class SchemaRefreshUtils {
-	@FunctionalInterface
-	interface ThrowingRunnable {
-		void run() throws Exception;
-	}
-
-	/**
-	 * Programmatically disables schema refreshes on the session and runs the provided Runnable,
-	 * taking care to restore the previous state of schema refresh config on the provided session.
-	 * Note that the session could have had schema refreshes enabled/disabled either
-	 * programmatically or via config.
-	 */
-	static void withDisabledSchema(Session session, ThrowingRunnable r) throws Exception {
-		boolean schemaEnabledPreviously = session.isSchemaMetadataEnabled();
-		session.setSchemaMetadataEnabled(false);
-		r.run();
-		session.setSchemaMetadataEnabled(null); // triggers schema refresh if results in true
-		if (schemaEnabledPreviously != session.isSchemaMetadataEnabled()) {
-			// user may have set it programmatically so set it back programmatically
-			session.setSchemaMetadataEnabled(schemaEnabledPreviously);
-		}
-	}
-}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SchemaUtils.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.cassandra.config;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+import com.datastax.oss.driver.api.core.session.Session;
+import com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures;
+
+/**
+ * Utility methods for executing schema actions.
+ *
+ * @author Ammar Khaku
+ * @author Mark Paluch
+ * @since 2.7
+ */
+class SchemaUtils {
+
+	/**
+	 * Programmatically disables schema refresh on the session and runs the provided {@link Runnable}. Takes care to
+	 * restore the previous state of schema refresh on the provided session. Note that the session could have had schema
+	 * refreshes enabled/disabled either programmatically or via config.
+	 *
+	 * @param session the session to use.
+	 * @param schemaAction the runnable code block.
+	 */
+	static void withSuspendedSchemaRefresh(Session session, Runnable schemaAction) {
+		CompletableFutures.getUninterruptibly(withSuspendedAsyncSchemaRefresh(session, schemaAction));
+	}
+
+	/**
+	 * Programmatically disables schema refresh on the session and runs the provided {@link Runnable}. Takes care to
+	 * restore the previous state of schema refresh on the provided session. Note that the session could have had schema
+	 * refreshes enabled/disabled either programmatically or via config.
+	 *
+	 * @param session the session to use.
+	 * @param schemaAction the runnable code block.
+	 * @return a {@link CompletionStage} providing a handle to the schema refresh completion.
+	 */
+	static CompletionStage<?> withSuspendedAsyncSchemaRefresh(Session session, Runnable schemaAction) {
+
+		boolean schemaEnabledPreviously = session.isSchemaMetadataEnabled();
+		if (schemaEnabledPreviously) {
+			session.setSchemaMetadataEnabled(false);
+		}
+
+		CompletionStage<?> schemaRefresh;
+
+		try {
+			schemaAction.run();
+		} finally {
+			if (schemaEnabledPreviously) {
+				// user may have set it programmatically so set it back programmatically
+				schemaRefresh = session.setSchemaMetadataEnabled(null);
+			} else {
+				schemaRefresh = CompletableFuture.completedFuture(null);
+			}
+		}
+
+		return schemaRefresh;
+	}
+
+	/**
+	 * Run a {@link Runnable} and refresh the schema after finishing the runnable.
+	 *
+	 * @param session the session to use.
+	 * @param schemaAction the runnable code block.
+	 */
+	static void withSchemaRefresh(Session session, Runnable schemaAction) {
+		CompletableFutures.getUninterruptibly(withAsyncSchemaRefresh(session, schemaAction));
+	}
+
+	/**
+	 * Run a {@link Runnable} and refresh the schema after finishing the runnable.
+	 *
+	 * @param session the session to use.
+	 * @param schemaAction the runnable code block.
+	 * @return a {@link CompletionStage} providing a handle to the schema refresh completion.
+	 */
+	static CompletionStage<?> withAsyncSchemaRefresh(Session session, Runnable schemaAction) {
+
+		schemaAction.run();
+
+		if (session.isSchemaMetadataEnabled()) {
+			return session.refreshSchemaAsync();
+		}
+
+		return CompletableFuture.completedFuture(null);
+	}
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SessionFactoryFactoryBean.java

@@ -56,6 +56,8 @@ public class SessionFactoryFactoryBean extends AbstractFactoryBean<SessionFactor
 
 	private SchemaAction schemaAction = SchemaAction.NONE;
 
+	private boolean suspendLifecycleSchemaRefresh = false;
+
 	/**
 	 * Set the {@link CassandraConverter} to use. Schema actions will derive table and user type information from the
 	 * {@link CassandraMappingContext} inside {@code converter}.
@@ -104,6 +106,24 @@ public class SessionFactoryFactoryBean extends AbstractFactoryBean<SessionFactor
 		this.schemaAction = schemaAction;
 	}
 
+	/**
+	 * Set whether to suspend schema refresh settings during {@link #afterPropertiesSet()} and {@link #destroy()}
+	 * lifecycle callbacks. Disabled by default to use schema metadata settings of the session configuration. When enabled
+	 * (set to {@code true}), then schema refresh during lifecycle methods is suspended until finishing schema actions to
+	 * avoid periodic schema refreshes for each DDL statement.
+	 * <p>
+	 * Suspending schema refresh can be useful to delay schema agreement until the entire schema is created. Note that
+	 * disabling schema refresh may interfere with schema actions. {@link SchemaAction#RECREATE_DROP_UNUSED} and
+	 * mapping-based schema creation rely on schema metadata.
+	 *
+	 * @param suspendLifecycleSchemaRefresh {@code true} to suspend the schema refresh during lifecycle callbacks;
+	 *          {@code false} otherwise to retain the session schema refresh configuration.
+	 * @since 2.7
+	 */
+	public void setSuspendLifecycleSchemaRefresh(boolean suspendLifecycleSchemaRefresh) {
+		this.suspendLifecycleSchemaRefresh = suspendLifecycleSchemaRefresh;
+	}
+
 	/**
 	 * Set the {@link CqlSession} to use.
 	 *
@@ -125,11 +145,19 @@ public class SessionFactoryFactoryBean extends AbstractFactoryBean<SessionFactor
 
 		super.afterPropertiesSet();
 
-		if (this.keyspacePopulator != null) {
-			this.keyspacePopulator.populate(getObject().getSession());
-		}
+		Runnable schemaActionRunnable = () -> {
+			if (this.keyspacePopulator != null) {
+				this.keyspacePopulator.populate(this.session);
+			}
 
-		SchemaRefreshUtils.withDisabledSchema(session, this::performSchemaAction);
+			performSchemaAction();
+		};
+
+		if (this.suspendLifecycleSchemaRefresh) {
+			SchemaUtils.withSuspendedSchemaRefresh(this.session, schemaActionRunnable);
+		} else {
+			SchemaUtils.withSchemaRefresh(this.session, schemaActionRunnable);
+		}
 	}
 
 	@Override
@@ -141,8 +169,16 @@ public class SessionFactoryFactoryBean extends AbstractFactoryBean<SessionFactor
 	@SuppressWarnings("all")
 	public void destroy() throws Exception {
 
-		if (this.keyspaceCleaner != null) {
-			this.keyspaceCleaner.populate(getObject().getSession());
+		Runnable schemaActionRunnable = () -> {
+			if (this.keyspaceCleaner != null) {
+				this.keyspaceCleaner.populate(this.session);
+			}
+		};
+
+		if (suspendLifecycleSchemaRefresh) {
+			SchemaUtils.withSuspendedAsyncSchemaRefresh(this.session, schemaActionRunnable);
+		} else {
+			schemaActionRunnable.run();
 		}
 	}
 
@@ -153,9 +189,9 @@ public class SessionFactoryFactoryBean extends AbstractFactoryBean<SessionFactor
 	}
 
 	/**
-	 * Perform the configure {@link SchemaAction} using {@link CassandraMappingContext} metadata.
+	 * Perform the configured {@link SchemaAction} using {@link CassandraMappingContext} metadata.
 	 */
-	protected void performSchemaAction() throws Exception {
+	protected void performSchemaAction() {
 
 		boolean create = false;
 		boolean drop = DEFAULT_DROP_TABLES;
@@ -190,22 +226,22 @@ public class SessionFactoryFactoryBean extends AbstractFactoryBean<SessionFactor
 	 * @param ifNotExists {@literal true} to perform creations fail-safe by adding {@code IF NOT EXISTS} to each creation
 	 *          statement.
 	 */
-	protected void createTables(boolean drop, boolean dropUnused, boolean ifNotExists) throws Exception {
+	protected void createTables(boolean drop, boolean dropUnused, boolean ifNotExists) {
 		performSchemaActions(drop, dropUnused, ifNotExists);
 	}
 
 	@SuppressWarnings("all")
-	private void performSchemaActions(boolean drop, boolean dropUnused, boolean ifNotExists) throws Exception {
+	private void performSchemaActions(boolean drop, boolean dropUnused, boolean ifNotExists) {
 
-		CassandraAdminOperations adminOperations = new CassandraAdminTemplate(getObject(), this.converter);
+		CassandraAdminOperations adminOperations = new CassandraAdminTemplate(this.session, this.converter);
 
-		CassandraPersistentEntitySchemaCreator schemaCreator =
-				new CassandraPersistentEntitySchemaCreator(this.converter.getMappingContext(), adminOperations);
+		CassandraPersistentEntitySchemaCreator schemaCreator = new CassandraPersistentEntitySchemaCreator(
+				this.converter.getMappingContext(), adminOperations);
 
 		if (drop) {
 
-			CassandraPersistentEntitySchemaDropper schemaDropper =
-					new CassandraPersistentEntitySchemaDropper(this.converter.getMappingContext(), adminOperations);
+			CassandraPersistentEntitySchemaDropper schemaDropper = new CassandraPersistentEntitySchemaDropper(
+					this.converter.getMappingContext(), adminOperations);
 
 			schemaDropper.dropTables(dropUnused);
 			schemaDropper.dropUserTypes(dropUnused);

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/SchemaActionIntegrationTests.java

@@ -26,6 +26,7 @@ import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.io.ByteArrayResource;
 import org.springframework.data.cassandra.core.cql.SessionCallback;
@@ -216,6 +217,16 @@ class SchemaActionIntegrationTests extends IntegrationTestsSupport {
 			return new ResourceKeyspacePopulator(new ByteArrayResource(CREATE_PERSON_TABLE_CQL.getBytes()));
 		}
 
+		@Bean
+		@Override
+		public SessionFactoryFactoryBean cassandraSessionFactory(CqlSession cqlSession) {
+
+			SessionFactoryFactoryBean bean = super.cassandraSessionFactory(cqlSession);
+			bean.setSuspendLifecycleSchemaRefresh(true);
+
+			return bean;
+		}
+
 		@Override
 		protected Set<Class<?>> getInitialEntitySet() throws ClassNotFoundException {
 			return Collections.singleton(Person.class);

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/SchemaRefreshUtilsUnitTests.java

@@ -1,66 +0,0 @@
-/*
- * Copyright 2022 the original author or authors.
- *
- * Licensed 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
- *
- *      https://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.springframework.data.cassandra.config;
-
-import static org.mockito.Mockito.*;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-import com.datastax.oss.driver.api.core.session.Session;
-
-/**
- * Test suite of unit tests testing the contract and functionality of the {@link SchemaRefreshUtils} class.
- */
-@ExtendWith(MockitoExtension.class)
-class SchemaRefreshUtilsUnitTests {
-	@Mock Session session;
-
-	@Test
-	void withDisabledSchemaRevert() throws Exception {
-		when(session.isSchemaMetadataEnabled()).thenReturn(true);
-		SchemaRefreshUtils.withDisabledSchema(session, () -> {});
-		verify(session).setSchemaMetadataEnabled(false);
-		verify(session).setSchemaMetadataEnabled(null);
-	}
-
-	@Test
-	void withDisabledSchemaDisabledPreviously() throws Exception {
-		when(session.isSchemaMetadataEnabled()).thenReturn(false);
-		SchemaRefreshUtils.withDisabledSchema(session, () -> {});
-		verify(session).setSchemaMetadataEnabled(false);
-		verify(session).setSchemaMetadataEnabled(null);
-	}
-
-	@Test
-	void withDisabledSchemaDisabledProgrammaticallyPreviously() throws Exception {
-		when(session.isSchemaMetadataEnabled()).thenReturn(false).thenReturn(true);
-		SchemaRefreshUtils.withDisabledSchema(session, () -> {});
-		verify(session, times(2)).setSchemaMetadataEnabled(false);
-		verify(session).setSchemaMetadataEnabled(null);
-	}
-
-	@Test
-	void withDisabledSchemaEnabledProgrammaticallyPreviously() throws Exception {
-		when(session.isSchemaMetadataEnabled()).thenReturn(true).thenReturn(false);
-		SchemaRefreshUtils.withDisabledSchema(session, () -> {});
-		verify(session).setSchemaMetadataEnabled(true);
-		verify(session).setSchemaMetadataEnabled(null);
-	}
-}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/SchemaUtilsUnitTests.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.data.cassandra.config;
+
+import static org.mockito.Mockito.*;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+import com.datastax.oss.driver.api.core.session.Session;
+
+/**
+ * Unit tests for {@link SchemaUtils}.
+ *
+ * @author Ammar Khaku
+ * @author Mark Paluch
+ */
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class SchemaUtilsUnitTests {
+
+	@Mock Session session;
+
+	@Test // GH-990, GH-1253
+	void shouldSuspendSchemaRefresh() {
+
+		when(session.isSchemaMetadataEnabled()).thenReturn(true);
+		when(session.setSchemaMetadataEnabled(true)).thenReturn(CompletableFuture.completedFuture(null));
+
+		SchemaUtils.withSuspendedAsyncSchemaRefresh(session, () -> {});
+
+		verify(session).setSchemaMetadataEnabled(false);
+		verify(session).setSchemaMetadataEnabled(null);
+	}
+
+	@Test // GH-990, GH-1253
+	void shouldRetainSchemaRefreshWhenSchemaMetadataDisabled() {
+
+		when(session.isSchemaMetadataEnabled()).thenReturn(false);
+
+		SchemaUtils.withSuspendedAsyncSchemaRefresh(session, () -> {});
+
+		verify(session, never()).setSchemaMetadataEnabled(anyBoolean());
+	}
+
+	@Test // GH-990, GH-1253
+	void shouldRefreshSchemaWhenSchemaMetadataEnabled() {
+
+		when(session.isSchemaMetadataEnabled()).thenReturn(true);
+		when(session.refreshSchemaAsync()).thenReturn(CompletableFuture.completedFuture(null));
+
+		SchemaUtils.withAsyncSchemaRefresh(session, () -> {});
+
+		verify(session, never()).setSchemaMetadataEnabled(anyBoolean());
+	}
+
+	@Test // GH-990, GH-1253
+	void shouldNotRefreshSchemaWhenSchemaMetadataDisabled() {
+
+		when(session.isSchemaMetadataEnabled()).thenReturn(false);
+
+		SchemaUtils.withAsyncSchemaRefresh(session, () -> {});
+
+		verify(session, never()).setSchemaMetadataEnabled(anyBoolean());
+		verify(session, never()).refreshSchemaAsync();
+	}
+
+}