GH-350: Upgrade SI Cassandra; re-enable tests (#381)

* GH-350: Upgrade SI Cassandra; re-enable tests Fixes https://github.com/spring-cloud/stream-applications/issues/350 * Use Testcontainers for Cassandra * Adjust components code style a bit * Convert `Book` testing entity to the `record` * Use `@DynamicPropertySource` to expose Docker ENV to Spring Boot configuration properties * Use switch expression for better code readability
2022-10-24 18:19:33 -04:00
parent 384d467002
commit 3d489991b1
11 changed files with 144 additions and 239 deletions
--- a/functions/consumer/cassandra-consumer/README.adoc
+++ b/functions/consumer/cassandra-consumer/README.adoc
@@ -29,4 +29,4 @@ See this link:src/test/java/org/springframework/cloud/fn/consumer/cassandra[test

 ## Other usage

-See this https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/cassandra-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application where it makes a Cassandra sink.
+See this https://github.com/spring-cloud/stream-applications/blob/main/applications/sink/cassandra-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application where it makes a Cassandra sink.
--- a/functions/consumer/cassandra-consumer/pom.xml
+++ b/functions/consumer/cassandra-consumer/pom.xml
@@ -14,11 +14,6 @@
 	<name>cassandra-consumer</name>
 	<description>Cassandra Consumer</description>

-	<properties>
-		<springIntegrationCassandara.version>0.8.0.RELEASE</springIntegrationCassandara.version>
-		<cassandra-unit-spring.version>4.3.1.0</cassandra-unit-spring.version>
-	</properties>
-
 	<dependencies>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
@@ -27,19 +22,11 @@
 		<dependency>
 			<groupId>org.springframework.integration</groupId>
 			<artifactId>spring-integration-cassandra</artifactId>
-			<version>${springIntegrationCassandara.version}</version>
 		</dependency>
 		<dependency>
-			<groupId>org.cassandraunit</groupId>
-			<artifactId>cassandra-unit-spring</artifactId>
-			<version>${cassandra-unit-spring.version}</version>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>cassandra</artifactId>
 			<scope>test</scope>
-			<exclusions>
-				<exclusion>
-					<groupId>com.addthis.metrics</groupId>
-					<artifactId>reporter-config3</artifactId>
-				</exclusion>
-			</exclusions>
 		</dependency>
 	</dependencies>

--- a/functions/consumer/cassandra-consumer/src/main/java/org/springframework/cloud/fn/consumer/cassandra/CassandraConsumerConfiguration.java
+++ b/functions/consumer/cassandra-consumer/src/main/java/org/springframework/cloud/fn/consumer/cassandra/CassandraConsumerConfiguration.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-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.
@@ -21,30 +21,34 @@ import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.function.Function;

+import com.datastax.oss.driver.api.core.ConsistencyLevel;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.util.StdDateFormat;
 import reactor.core.publisher.Mono;

 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.fn.consumer.cassandra.cluster.CassandraAppClusterConfiguration;
 import org.springframework.cloud.fn.consumer.cassandra.query.ColumnNameExtractor;
 import org.springframework.cloud.fn.consumer.cassandra.query.InsertQueryColumnNameExtractor;
 import org.springframework.cloud.fn.consumer.cassandra.query.UpdateQueryColumnNameExtractor;
 import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 import org.springframework.data.cassandra.core.InsertOptions;
 import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
 import org.springframework.data.cassandra.core.UpdateOptions;
 import org.springframework.data.cassandra.core.WriteResult;
 import org.springframework.data.cassandra.core.cql.WriteOptions;
+import org.springframework.integration.JavaUtils;
 import org.springframework.integration.cassandra.outbound.CassandraMessageHandler;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
-import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.support.json.Jackson2JsonObjectMapper;
 import org.springframework.integration.transformer.AbstractPayloadTransformer;
 import org.springframework.messaging.MessageHandler;
@@ -56,8 +60,9 @@ import org.springframework.util.StringUtils;
 * @author Ashu Gairola
 * @author Akos Ratku
 */
-@Configuration
+@AutoConfiguration
@EnableConfigurationProperties(CassandraConsumerProperties.class)
+@Import(CassandraAppClusterConfiguration.class)
 public class CassandraConsumerConfiguration {

 	@Autowired
@@ -65,12 +70,13 @@ public class CassandraConsumerConfiguration {

 	@Bean
 	public IntegrationFlow cassandraConsumerFlow(MessageHandler cassandraSinkMessageHandler,
-												ObjectMapper objectMapper) {
-		IntegrationFlowBuilder integrationFlowBuilder =
-				IntegrationFlows.from(CassandraConsumerFunction.class);
-		if (StringUtils.hasText(this.cassandraSinkProperties.getIngestQuery())) {
+			ObjectMapper objectMapper) {
+
+		IntegrationFlowBuilder integrationFlowBuilder = IntegrationFlow.from(CassandraConsumerFunction.class);
+		String ingestQuery = this.cassandraSinkProperties.getIngestQuery();
+		if (StringUtils.hasText(ingestQuery)) {
 			integrationFlowBuilder.transform(
-					new PayloadToMatrixTransformer(objectMapper, this.cassandraSinkProperties.getIngestQuery(),
+					new PayloadToMatrixTransformer(objectMapper, ingestQuery,
 							CassandraMessageHandler.Type.UPDATE == this.cassandraSinkProperties.getQueryType()
 									? new UpdateQueryColumnNameExtractor()
 									: new InsertQueryColumnNameExtractor()));
@@ -82,43 +88,35 @@ public class CassandraConsumerConfiguration {

 	@Bean
 	public MessageHandler cassandraSinkMessageHandler(ReactiveCassandraOperations cassandraOperations) {
-		CassandraMessageHandler cassandraMessageHandler =
-				this.cassandraSinkProperties.getQueryType() != null
-						? new CassandraMessageHandler(cassandraOperations, this.cassandraSinkProperties.getQueryType())
-						: new CassandraMessageHandler(cassandraOperations);
+		CassandraMessageHandler.Type queryType =
+				Optional.ofNullable(this.cassandraSinkProperties.getQueryType())
+						.orElse(CassandraMessageHandler.Type.INSERT);
+
+		CassandraMessageHandler cassandraMessageHandler = new CassandraMessageHandler(cassandraOperations, queryType);
 		cassandraMessageHandler.setProducesReply(true);
-		cassandraMessageHandler.setAsync(true);
-		if (this.cassandraSinkProperties.getConsistencyLevel() != null
-				|| this.cassandraSinkProperties.getTtl() > 0) {
+		int ttl = this.cassandraSinkProperties.getTtl();
+		ConsistencyLevel consistencyLevel = this.cassandraSinkProperties.getConsistencyLevel();
+		if (consistencyLevel != null || ttl > 0) {

-			WriteOptions.WriteOptionsBuilder writeOptionsBuilder = WriteOptions.builder();
+			WriteOptions.WriteOptionsBuilder writeOptionsBuilder =
+					switch (queryType) {
+						case INSERT -> InsertOptions.builder();
+						case UPDATE -> UpdateOptions.builder();
+						default -> WriteOptions.builder();
+					};

-			switch (this.cassandraSinkProperties.getQueryType()) {
-
-				case INSERT:
-					writeOptionsBuilder = InsertOptions.builder();
-					break;
-				case UPDATE:
-					writeOptionsBuilder = UpdateOptions.builder();
-					break;
-			}
-
-			if (this.cassandraSinkProperties.getConsistencyLevel() != null) {
-				writeOptionsBuilder.consistencyLevel(this.cassandraSinkProperties.getConsistencyLevel());
-			}
-
-			if (this.cassandraSinkProperties.getTtl() > 0) {
-				writeOptionsBuilder.ttl(this.cassandraSinkProperties.getTtl());
-			}
+			JavaUtils.INSTANCE
+					.acceptIfNotNull(consistencyLevel, writeOptionsBuilder::consistencyLevel)
+					.acceptIfCondition(ttl > 0, ttl, writeOptionsBuilder::ttl);

 			cassandraMessageHandler.setWriteOptions(writeOptionsBuilder.build());
 		}
-		if (StringUtils.hasText(this.cassandraSinkProperties.getIngestQuery())) {
-			cassandraMessageHandler.setIngestQuery(this.cassandraSinkProperties.getIngestQuery());
-		}
-		else if (this.cassandraSinkProperties.getStatementExpression() != null) {
-			cassandraMessageHandler.setStatementExpression(this.cassandraSinkProperties.getStatementExpression());
-		}
+
+		JavaUtils.INSTANCE
+				.acceptIfHasText(this.cassandraSinkProperties.getIngestQuery(), cassandraMessageHandler::setIngestQuery)
+				.acceptIfNotNull(this.cassandraSinkProperties.getStatementExpression(),
+						cassandraMessageHandler::setStatementExpression);
+
 		return cassandraMessageHandler;
 	}

@@ -164,8 +162,7 @@ public class CassandraConsumerConfiguration {
 						List<Object> row = new ArrayList<>(this.columns.size());
 						for (String column : this.columns) {
 							Object value = entity.get(column);
-							if (value instanceof String) {
-								String string = (String) value;
+							if (value instanceof String string) {
 								if (this.dateFormat.looksLikeISO8601(string)) {
 									synchronized (this.dateFormat) {
 										value = new Date(this.dateFormat.parse(string).getTime()).toLocalDate();
--- a/functions/consumer/cassandra-consumer/src/main/java/org/springframework/cloud/fn/consumer/cassandra/cluster/CassandraAppClusterConfiguration.java
+++ b/functions/consumer/cassandra-consumer/src/main/java/org/springframework/cloud/fn/consumer/cassandra/cluster/CassandraAppClusterConfiguration.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-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.
@@ -47,7 +47,6 @@ import org.springframework.core.env.Environment;
 import org.springframework.core.type.AnnotationMetadata;
 import org.springframework.data.cassandra.config.CqlSessionFactoryBean;
 import org.springframework.data.cassandra.core.ReactiveCassandraTemplate;
-import org.springframework.data.cassandra.core.cql.CqlTemplate;
 import org.springframework.data.cassandra.core.cql.ReactiveCqlOperations;
 import org.springframework.data.cassandra.core.cql.generator.CreateKeyspaceCqlGenerator;
 import org.springframework.data.cassandra.core.cql.keyspace.CreateKeyspaceSpecification;
@@ -75,9 +74,9 @@ public class CassandraAppClusterConfiguration {
 							try {
 								builder.withSslContext(TrustAllSSLContextFactory.getSslContext());
 							}
-							catch (NoSuchAlgorithmException | KeyManagementException e) {
+							catch (NoSuchAlgorithmException | KeyManagementException ex) {
 								throw new BeanInitializationException(
-										"Unable to configure a Cassandra cluster using SSL.", e);
+										"Unable to configure a Cassandra cluster using SSL.", ex);
 							}

 						});
@@ -93,11 +92,9 @@ public class CassandraAppClusterConfiguration {
 						.ifNotExists();

 		String createKeySpaceQuery = new CreateKeyspaceCqlGenerator(createKeyspaceSpecification).toCql();
-		CqlSession systemSession =
-				cqlSessionBuilder.withKeyspace(CqlSessionFactoryBean.CASSANDRA_SYSTEM_SESSION).build();
-
-		CqlTemplate template = new CqlTemplate(systemSession);
-		template.execute(createKeySpaceQuery);
+		try (var systemSession = cqlSessionBuilder.withKeyspace(CqlSessionFactoryBean.CASSANDRA_SYSTEM_SESSION).build()) {
+			systemSession.execute(createKeySpaceQuery);
+		}

 		return null;
 	}
@@ -117,7 +114,7 @@ public class CassandraAppClusterConfiguration {

 		String scripts =
 				new Scanner(cassandraClusterProperties.getInitScript().getInputStream(),
-						StandardCharsets.UTF_8.name())
+						StandardCharsets.UTF_8)
 						.useDelimiter("\\A")
 						.next();

--- a/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraConsumerApplicationTests.java
+++ b/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraConsumerApplicationTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-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.
@@ -19,41 +19,35 @@ package org.springframework.cloud.fn.consumer.cassandra;
 import java.time.LocalDate;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.function.Function;

-import org.cassandraunit.spring.CassandraUnitDependencyInjectionIntegrationTestExecutionListener;
-import org.cassandraunit.spring.EmbeddedCassandra;
-import org.cassandraunit.utils.EmbeddedCassandraServerHelper;
-import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Disabled;
+
 import reactor.core.publisher.Mono;

 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.cloud.fn.consumer.cassandra.domain.Book;
+import org.springframework.context.annotation.Import;
 import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.core.WriteResult;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.TestExecutionListeners;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;

 /**
 * @author Artem Bilan
 */
-@Disabled
-@TestExecutionListeners(mergeMode = TestExecutionListeners.MergeMode.MERGE_WITH_DEFAULTS,
-		listeners = CassandraUnitDependencyInjectionIntegrationTestExecutionListener.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
 		properties = {
-				"spring.data.cassandra.keyspaceName=" + CassandraConsumerApplicationTests.CASSANDRA_KEYSPACE,
-				"spring.data.cassandra.localDatacenter=datacenter1",
+				"spring.cassandra.keyspace-name=" + CassandraConsumerApplicationTests.CASSANDRA_KEYSPACE,
 				"cassandra.cluster.createKeyspace=true" })
-@EmbeddedCassandra(configuration = EmbeddedCassandraServerHelper.CASSANDRA_RNDPORT_YML_FILE, timeout = 120000)
@DirtiesContext
-abstract class CassandraConsumerApplicationTests {
+abstract class CassandraConsumerApplicationTests implements CassandraContainerTest {

 	static final String CASSANDRA_KEYSPACE = "test";

@@ -63,16 +57,14 @@ abstract class CassandraConsumerApplicationTests {
 	@Autowired
 	protected Function<Object, Mono<? extends WriteResult>> cassandraConsumer;

-	@BeforeAll
-	static void setUp() {
-		EmbeddedCassandraServerHelper.getSession();
-		System.setProperty("spring.data.cassandra.contactPoints",
-				EmbeddedCassandraServerHelper.getHost() + ':' + EmbeddedCassandraServerHelper.getNativeTransportPort());
-	}

-	@AfterAll
-	static void cleanup() {
-		System.clearProperty("spring.data.cassandra.contactPoints");
+	@DynamicPropertySource
+	static void registerConfigurationProperties(DynamicPropertyRegistry registry) {
+		registry.add("spring.cassandra.localDatacenter", () -> CASSANDRA_CONTAINER.getLocalDatacenter());
+		registry.add("spring.cassandra.contactPoints", () ->
+			Optional.of(CASSANDRA_CONTAINER.getContactPoint())
+					.map(contactPoint -> contactPoint.getAddress().getHostAddress() + ':' + contactPoint.getPort())
+					.get());
 	}

 	@AfterEach
@@ -83,23 +75,23 @@ abstract class CassandraConsumerApplicationTests {
 	protected static List<Book> getBookList(int numBooks) {

 		List<Book> books = new ArrayList<>();
-
-		Book b;
 		for (int i = 0; i < numBooks; i++) {
-			b = new Book();
-			b.setIsbn(UUID.randomUUID());
-			b.setTitle("Spring Cloud Data Flow Guide");
-			b.setAuthor("SCDF Guru");
-			b.setPages(i * 10 + 5);
-			b.setInStock(true);
-			b.setSaleDate(LocalDate.now());
-			books.add(b);
+			books.add(
+					new Book(
+							UUID.randomUUID(),
+							"Spring Cloud Data Flow Guide",
+							"SCDF Guru",
+							i * 10 + 5,
+							LocalDate.now(),
+							true));
 		}

 		return books;
 	}

-	@SpringBootApplication
+	@SpringBootConfiguration
+	@EnableAutoConfiguration
+	@Import(CassandraConsumerConfiguration.class)
 	static class CassandraConsumerTestApplication {

 	}
--- a/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraContainerTest.java
+++ b/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraContainerTest.java
@@ -0,0 +1,45 @@
+/*
+ * 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.cloud.fn.consumer.cassandra;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.testcontainers.containers.CassandraContainer;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+/**
+ * The base contract for JUnit tests based on the container for Apache Cassandra.
+ * The Testcontainers 'reuse' option must be disabled,so, Ryuk container is started
+ * and will clean all the containers up from this test suite after JVM exit.
+ * Since the MqSQL container instance is shared via static property, it is going to be
+ * started only once per JVM, therefore the target Docker container is reused automatically.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+@Testcontainers(disabledWithoutDocker = true)
+public interface CassandraContainerTest {
+
+	CassandraContainer<?> CASSANDRA_CONTAINER = new CassandraContainer<>("cassandra:4.1");
+
+	@BeforeAll
+	static void startContainer() {
+		System.setProperty("datastax-java-driver.basic.request.timeout", "10 seconds");
+		CASSANDRA_CONTAINER.start();
+	}
+
+}
--- a/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraEntityInsertTests.java
+++ b/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraEntityInsertTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-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.
@@ -19,7 +19,6 @@ package org.springframework.cloud.fn.consumer.cassandra;
 import java.time.LocalDate;
 import java.util.UUID;

-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -33,22 +32,21 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
 * @author Artem Bilan
 */
-@Disabled
@TestPropertySource(properties = {
-		"spring.data.cassandra.schema-action=RECREATE",
+		"spring.cassandra.schema-action=RECREATE",
 		"cassandra.cluster.entity-base-packages=org.springframework.cloud.fn.consumer.cassandra.domain" })
 class CassandraEntityInsertTests extends CassandraConsumerApplicationTests {

 	@Test
-	@Disabled
 	void testInsert() {
-		Book book = new Book();
-		book.setIsbn(UUID.randomUUID());
-		book.setTitle("Spring Integration Cassandra");
-		book.setAuthor("Cassandra Guru");
-		book.setPages(521);
-		book.setSaleDate(LocalDate.now());
-		book.setInStock(true);
+		Book book =
+				new Book(
+						UUID.randomUUID(),
+						"Spring Integration Cassandra",
+						"Cassandra Guru",
+						521,
+						LocalDate.now(),
+						true);

 		Mono<? extends WriteResult> result = this.cassandraConsumer.apply(book);

--- a/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraIngestInsertTests.java
+++ b/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraIngestInsertTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-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.
@@ -19,7 +19,6 @@ package org.springframework.cloud.fn.consumer.cassandra;
 import java.util.List;

 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -35,7 +34,6 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
 * @author Artem Bilan
 */
-@Disabled
@TestPropertySource(properties = {
 		"cassandra.cluster.init-script=init-db.cql",
 		"cassandra.ingest-query=" +
--- a/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraIngestNamedParamsTests.java
+++ b/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraIngestNamedParamsTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-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.
@@ -19,7 +19,6 @@ package org.springframework.cloud.fn.consumer.cassandra;
 import java.util.List;

 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -36,7 +35,6 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
 * @author Artem Bilan
 */
-@Disabled
@TestPropertySource(properties = {
 		"cassandra.cluster.init-script=init-db.cql",
 		"cassandra.ingest-query=" +
--- a/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraIngestUpdateTests.java
+++ b/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/CassandraIngestUpdateTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-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.
@@ -19,7 +19,6 @@ package org.springframework.cloud.fn.consumer.cassandra;
 import java.util.List;

 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -35,7 +34,6 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
 * @author Artem Bilan
 */
-@Disabled
@TestPropertySource(properties = {
 		"cassandra.cluster.init-script=init-db.cql",
 		"cassandra.ingest-query=" +
--- a/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/domain/Book.java
+++ b/functions/consumer/cassandra-consumer/src/test/java/org/springframework/cloud/fn/consumer/cassandra/domain/Book.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-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.
@@ -19,6 +19,7 @@ package org.springframework.cloud.fn.consumer.cassandra.domain;
 import java.time.LocalDate;
 import java.util.UUID;

+import org.springframework.data.cassandra.core.mapping.Indexed;
 import org.springframework.data.cassandra.core.mapping.PrimaryKey;
 import org.springframework.data.cassandra.core.mapping.Table;

@@ -29,118 +30,12 @@ import org.springframework.data.cassandra.core.mapping.Table;
 * @author Artem Bilan
 */
@Table("book")
-public class Book {
-
-	@PrimaryKey
-	private UUID isbn;
-
-	private String title;
-
-	private String author;
-
-	private int pages;
-
-	private LocalDate saleDate;
-
-	private boolean inStock;
-
-	public Book() {
-	}
-
-	public Book(UUID isbn, String title, String author) {
-		this.isbn = isbn;
-		this.title = title;
-		this.author = author;
-	}
-
-	/**
-	 * @return Returns the isbn.
-	 */
-	public UUID getIsbn() {
-		return this.isbn;
-	}
-
-	/**
-	 * @return Returns the saleDate.
-	 */
-	public LocalDate getSaleDate() {
-		return this.saleDate;
-	}
-
-	/**
-	 * @param saleDate The saleDate to set.
-	 */
-	public void setSaleDate(LocalDate saleDate) {
-		this.saleDate = saleDate;
-	}
-
-	/**
-	 * @return Returns the inStock.
-	 */
-	public boolean isInStock() {
-		return this.inStock;
-	}
-
-	/**
-	 * @param inStock The isInStock to set.
-	 */
-	public void setInStock(boolean inStock) {
-		this.inStock = inStock;
-	}
-
-	/**
-	 * @param isbn The isbn to set.
-	 */
-	public void setIsbn(UUID isbn) {
-		this.isbn = isbn;
-	}
-
-	/**
-	 * @return Returns the title.
-	 */
-	public String getTitle() {
-		return this.title;
-	}
-
-	/**
-	 * @param title The title to set.
-	 */
-	public void setTitle(String title) {
-		this.title = title;
-	}
-
-	/**
-	 * @return Returns the author.
-	 */
-	public String getAuthor() {
-		return this.author;
-	}
-
-	/**
-	 * @param author The author to set.
-	 */
-	public void setAuthor(String author) {
-		this.author = author;
-	}
-
-	/**
-	 * @return Returns the pages.
-	 */
-	public int getPages() {
-		return this.pages;
-	}
-
-	/**
-	 * @param pages The pages to set.
-	 */
-	public void setPages(int pages) {
-		this.pages = pages;
-	}
-
-	@Override
-	public String toString() {
-		return ("isbn -> " + this.isbn) + "\n" + "tile -> " + this.title + "\n" + "author -> " + this.author
-				+ "\n" + "pages -> " + this.pages + "\n";
-	}
+public record Book(
+		@PrimaryKey UUID isbn,
+		String title,
+		@Indexed String author,
+		Integer pages,
+		LocalDate saleDate,
+		Boolean isInStock) {

 }