Allow lenient infrastructure startup.

By avoiding interaction with a potential lazy proxy for a given CqlSession until it becomes mandatory we allow users to leverage lazy bean intialization for CqlSession during application startup.
To fully enable this changes we need changes in the boot autoconfigration.

Closes #380
Original pull request: #1485

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

@@ -145,6 +145,10 @@ public class SessionFactoryFactoryBean extends AbstractFactoryBean<SessionFactor
 
 		super.afterPropertiesSet();
 
+		if(!shouldRunSchemaAction()) {
+			return;
+		}
+
 		Runnable schemaActionRunnable = () -> {
 			if (this.keyspacePopulator != null) {
 				this.keyspacePopulator.populate(this.session);
@@ -169,6 +173,10 @@ public class SessionFactoryFactoryBean extends AbstractFactoryBean<SessionFactor
 	@SuppressWarnings("all")
 	public void destroy() throws Exception {
 
+		if(!shouldRunSchemaAction()) {
+			return;
+		}
+
 		Runnable schemaActionRunnable = () -> {
 			if (this.keyspaceCleaner != null) {
 				this.keyspaceCleaner.populate(this.session);
@@ -230,6 +238,14 @@ public class SessionFactoryFactoryBean extends AbstractFactoryBean<SessionFactor
 		performSchemaActions(drop, dropUnused, ifNotExists);
 	}
 
+	/**
+	 * @return {@literal true} if schema action or {@link KeyspacePopulator} defined.
+	 * @since 4.3
+	 */
+	protected boolean shouldRunSchemaAction() {
+		return keyspacePopulator != null || SchemaAction.NONE != this.schemaAction;
+	}
+
 	@SuppressWarnings("all")
 	private void performSchemaActions(boolean drop, boolean dropUnused, boolean ifNotExists) {
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java

@@ -24,6 +24,7 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -61,6 +62,7 @@ import org.springframework.data.mapping.model.ValueExpressionParameterValueProvi
 import org.springframework.data.projection.EntityProjection;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
+import org.springframework.data.util.Lazy;
 import org.springframework.data.util.Predicates;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
@@ -98,7 +100,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 	private final CassandraMappingContext mappingContext;
 
-	private CodecRegistry codecRegistry;
+	private Supplier<CodecRegistry> codecRegistry;
 
 	private @Nullable UserTypeResolver userTypeResolver;
 
@@ -234,8 +236,20 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	public void setCodecRegistry(CodecRegistry codecRegistry) {
 
 		Assert.notNull(codecRegistry, "CodecRegistry must not be null");
+		setCodecRegistry(() -> codecRegistry);
+	}
 
-		this.codecRegistry = codecRegistry;
+	/**
+	 * Sets the {@link Supplier} used for obtaining the {@link CodecRegistry} to use.
+	 *
+	 * @param codecRegistry must not be {@literal null}.
+	 * @since 4.3
+	 */
+	public void setCodecRegistry(Supplier<CodecRegistry> codecRegistry) {
+
+		Assert.notNull(codecRegistry, "CodecRegistry provider must not be null");
+
+		this.codecRegistry = Lazy.of(codecRegistry);
 	}
 
 	/**
@@ -245,8 +259,11 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	 * @since 3.0
 	 */
 	@Override
+	@SuppressWarnings({ "deprecation" })
 	public CodecRegistry getCodecRegistry() {
-		return this.codecRegistry != null ? this.codecRegistry : getMappingContext().getCodecRegistry();
+
+		CodecRegistry registry = this.codecRegistry != null ? this.codecRegistry.get() : null;
+		return  registry != null ? registry : getMappingContext().getCodecRegistry();
 	}
 
 	/**

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

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2024 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.assertj.core.api.Assertions.*;
+
+import java.net.InetSocketAddress;
+import java.util.Collections;
+
+import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.internal.core.metadata.DefaultEndPoint;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.data.cassandra.SessionFactory;
+import org.springframework.data.cassandra.core.CassandraTemplate;
+import org.springframework.data.cassandra.core.Person;
+import org.springframework.data.cassandra.core.convert.CassandraConverter;
+import org.springframework.data.cassandra.core.convert.CassandraCustomConversions;
+import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
+import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
+import org.springframework.data.cassandra.core.mapping.SimpleUserTypeResolver;
+
+/**
+ * @author Christoph Strobl
+ */
+class LazyStartupConfigurationTest {
+
+	@Test // GH-380
+	void shouldDelayCqlSessionBeanInitializationTillFirstUsage() {
+
+		GenericApplicationContext ctx = new AnnotationConfigApplicationContext(LazyStartupConfig.class);
+
+		CassandraTemplate template = ctx.getBean(CassandraTemplate.class);
+		assertThat(template).isNotNull();
+		assertThatExceptionOfType(BeanCreationException.class).isThrownBy(() -> template.count(Person.class));
+	}
+
+	@Configuration
+	static class LazyStartupConfig {
+
+		@Lazy
+		@Bean
+		CqlSession cqlSession() {
+
+			return CqlSession.builder()
+					.addContactEndPoint(new DefaultEndPoint(InetSocketAddress.createUnresolved("127.0.0.2", 9042)))
+					.withKeyspace("system")
+					.build();
+		}
+
+		@Bean
+		public SessionFactoryFactoryBean cassandraSessionFactory(CassandraConverter converter, @Lazy CqlSession cqlSession) {
+
+			SessionFactoryFactoryBean session = new SessionFactoryFactoryBean();
+			session.setSession(cqlSession);
+			session.setConverter(converter);
+			session.setSchemaAction(SchemaAction.NONE);
+			return session;
+		}
+
+		@Bean
+		public CassandraMappingContext cassandraMappingContext() {
+			CassandraMappingContext context = new CassandraMappingContext();
+			context.setSimpleTypeHolder(new CassandraCustomConversions(Collections.emptyList()).getSimpleTypeHolder());
+			return context;
+		}
+
+		@Bean
+		public CassandraConverter cassandraConverter(CassandraMappingContext mappingContext, @Lazy CqlSession cqlSession) {
+
+			MappingCassandraConverter converter = new MappingCassandraConverter(mappingContext);
+			converter.setCodecRegistry(() -> cqlSession.getContext().getCodecRegistry());
+			converter.setCustomConversions(mappingContext.getCustomConversions());
+
+			CqlIdentifier keyspace = CqlIdentifier.fromCql("system");
+			converter.setUserTypeResolver(new SimpleUserTypeResolver(cqlSession::getMetadata, keyspace));
+			return converter;
+		}
+
+		@Bean
+		public CassandraTemplate cassandraTemplate(SessionFactory sessionFactory, CassandraConverter converter) {
+			return new CassandraTemplate(sessionFactory, converter);
+		}
+	}
+}

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

@@ -49,7 +49,7 @@ import com.datastax.oss.driver.api.core.metadata.schema.TableMetadata;
  * @author Mark Paluch
  * @see AbstractKeyspaceCreatingIntegrationTests
  */
-class SchemaActionIntegrationTests extends IntegrationTestsSupport {
+class SchemaActionIntegrationTests extends AbstractKeyspaceCreatingIntegrationTests {
 
 	private static final String CREATE_PERSON_TABLE_CQL = "CREATE TABLE IF NOT EXISTS person (id int, firstName text, lastName text, PRIMARY KEY(id));";
 
@@ -66,7 +66,7 @@ class SchemaActionIntegrationTests extends IntegrationTestsSupport {
 	protected <T> T doInSessionWithConfiguration(Class<?> annotatedClass, SessionCallback<T> sessionCallback) {
 
 		try (ConfigurableApplicationContext applicationContext = newApplicationContext(annotatedClass)) {
-			return sessionCallback.doInSession(applicationContext.getBean(CqlSession.class));
+			return sessionCallback.doInSession(getSession());
 		}
 	}
 
@@ -154,6 +154,21 @@ class SchemaActionIntegrationTests extends IntegrationTestsSupport {
 		});
 	}
 
+	@Test // GH-380
+	void shouldDoNotingIfSchemaActionIsNoneAndNoPopulateScriptPresent() {
+
+		doInSessionWithConfiguration(CreateWithoutSchemaAction.class, session -> {
+
+			KeyspaceMetadata keyspaceMetadata = session.refreshSchema().getKeyspace(session.getKeyspace().get()).orElse(null);
+
+			assertThat(keyspaceMetadata).isNotNull();
+			assertThat(keyspaceMetadata.getTables()).isEmpty();
+
+			return null;
+		});
+
+	}
+
 	@Configuration
 	static class CreateWithNoExistingTableConfiguration extends IntegrationTestConfig {
 
@@ -168,6 +183,20 @@ class SchemaActionIntegrationTests extends IntegrationTestsSupport {
 		}
 	}
 
+	@Configuration
+	static class CreateWithoutSchemaAction extends IntegrationTestConfig {
+
+		@Override
+		public SchemaAction getSchemaAction() {
+			return SchemaAction.NONE;
+		}
+
+		@Override
+		protected KeyspacePopulator keyspacePopulator() {
+			return null;
+		}
+	}
+
 	@Configuration
 	static class CreateWithExistingTableConfiguration extends IntegrationTestConfig {