diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java index dff59a893..4779de7f2 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java +++ b/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> 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. + *

+ * 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> 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(); } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SchemaRefreshUtils.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SchemaRefreshUtils.java deleted file mode 100644 index beb5c144d..000000000 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SchemaRefreshUtils.java +++ /dev/null @@ -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); - } - } -} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SchemaUtils.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SchemaUtils.java new file mode 100644 index 000000000..388c30b06 --- /dev/null +++ b/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); + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SessionFactoryFactoryBean.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SessionFactoryFactoryBean.java index 6e5fb8e76..246956963 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SessionFactoryFactoryBean.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SessionFactoryFactoryBean.java @@ -56,6 +56,8 @@ public class SessionFactoryFactoryBean extends AbstractFactoryBean + * 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 { + 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 { + 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> getInitialEntitySet() throws ClassNotFoundException { return Collections.singleton(Person.class); diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/SchemaRefreshUtilsUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/SchemaRefreshUtilsUnitTests.java deleted file mode 100644 index ecef2ab33..000000000 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/SchemaRefreshUtilsUnitTests.java +++ /dev/null @@ -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); - } -} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/SchemaUtilsUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/SchemaUtilsUnitTests.java new file mode 100644 index 000000000..2e31ff2d8 --- /dev/null +++ b/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(); + } + +}