当 运行 使用 Spring Cloud Stream、Spring Data Cassandra 和 Cassandra-Unit-Spring 进行集成测试时，怪异随之而来
Weirdness ensues when running integraton tests with Spring Cloud Stream, Spring Data Cassandra and Cassandra-Unit-Spring
我将 Cassandra 用作我的应用程序的事件存储，Spring Cloud Stream 接收事件并 Spring Cloud Cassandra 保存它们。
为了监听不同的事件，我创建了一个具有两种用户定义类型的 ClientEvent：客户端和产品（类 都用 @UserDefinedType 注释），如下所示：
@Table
public class ClientEvent {
@PrimaryKey
@CassandraType(type = DataType.Name.UUID)
private String id; // created with UUIDs.timeBased().toString()
@CassandraType(type = DataType.Name.UDT, userTypeName = "client")
private Client client; // has an email field
@CassandraType(type = DataType.Name.UDT, userTypeName = "product")
private Product product; // has name and cost fields
}
我的 EventConsumer 看起来像这样，并且适用于不同的事件：
@EnableBinding(Sink.class)
public class EventConsumer {
@Inject
private EventRepository eventRepository;
@Override
@StreamListener(Sink.INPUT)
public void clientEvent(ClientEvent event) {
eventRepository.save(event);
}
}
我还有一个控制器 returns 给定客户端的所有事件（所述控制器的代码非常标准，所以我不会在这里 post 它）。
为了测试所有这些，我设置了一些集成测试 spring-cloud-contract-stub-运行ner 和 cassandra-unit-spring 到运行 嵌入式卡桑德拉。我只有以下配置，在我的 application.yml:
spring:
profiles: test
data:
cassandra:
keyspace-name: testkeyspace
contact-points: localhost
port: 9142
这是 Spock 测试：
@SpringBootTest(classes = EventsApplication.class, webEnvironment = RANDOM_PORT)
@AutoConfigureStubRunner(ids = ["example.com:clients-service",
"example.com:products-service"], stubsMode = StubsMode.LOCAL)
@TestExecutionListeners(listeners = [
CassandraUnitDependencyInjectionTestExecutionListener.class,
DependencyInjectionTestExecutionListener.class
])
@CassandraDataSet(keyspace = "testkeyspace", value = "dataset.cql")
@EmbeddedCassandra(timeout = 60000L)
@ActiveProfiles("test")
class EventsIntegrationTests extends Specification {
@Inject
StubTrigger stubTrigger
@Inject
private TestRestTemplate restTemplate
def "client created"() {
given: "that a client has been created"
stubTrigger.trigger("client_created") //stub defined in clients-service
when: "I fetch the events for the current client"
def response = restTemplate.getForEntity("/", ClientEvent[])
then: "it should be a clientCreated event"
ClientEvent event = response.body[0]
event.name == "clientCreated"
event.client == new Client("test") // hashCode and equals implemented
}
def "product bought"() {
given: "that a product has been bought"
stubTrigger.trigger("product_bought") //stub defined in products-service
}
when: "I fetch the events for the current client"
def response = restTemplate.getForEntity("/", ClientEvent[])
then: "it should be a productBought event"
ClientEvent event = response.body[0]
event.name == "productBought"
event.client == new Client("test") // hashCode and equals implemented
event.product == new Product("Lamp", 100D)
}
}
dataset.cql 文件创建测试密钥空间、客户端和产品类型，然后是 ClientEvent table。到目前为止，一切都很好。问题是，虽然第一个测试 运行 没问题，但第二个测试在 EventConsumer 中保存事件时给了我这个 NullPointerException:
org.springframework.messaging.MessagingException: Exception thrown while invoking com.example.events.consumers.EventConsumer#clientEvent[1 args]; nested exception is java.lang.NullPointerException
[...]
Caused by: java.lang.NullPointerException: null
at org.springframework.data.cassandra.core.mapping.SimpleUserTypeResolver.resolveType(SimpleUserTypeResolver.java:63) ~[spring-data-cassandra-2.0.6.RELEASE.jar:2.0.6.RELEASE]
我发现如果我在每次测试之间调试或等待一秒钟（在 setup() 方法中使用 Thread.sleep(1000)），我的测试 运行 很好！我注意到在每次测试之间，cassandra-unit-spring 删除并重新创建数据库。在我看来，第二次测试发生在重新创建用户定义类型之前，这就是它抛出 NullPointerException 的原因：
2018-05-31 22:43:43.552 INFO 14976 --- [port-Requests-1] o.a.cassandra.service.MigrationManager : Drop Keyspace 'testkeyspace'
2018-05-31 22:43:53.445 INFO 14976 --- [igrationStage:1] o.a.cassandra.utils.memory.BufferPool : Global buffer pool is enabled, when pool is exhausted (max is 512.000MiB) it will allocate on heap
2018-05-31 22:43:53.511 INFO 14976 --- [port-Requests-3] o.a.cassandra.service.MigrationManager : Create new Keyspace: KeyspaceMetadata{name=testkeyspace, params=KeyspaceParams{durable_writes=false, replication=ReplicationParams{class=org.apache.cassandra.locator.SimpleStrategy, replication_factor=1}}, tables=[], views=[], functions=[], types=[]}
2018-05-31 22:43:53.603 INFO 14976 --- [port-Requests-2] o.a.cassandra.service.MigrationManager : Create new table: org.apache.cassandra.config.CFMetaData@1f5471ca[cfId=3d9ed530-653d-11e8-b838-27af117b0453,ksName=testkeyspace,cfName=clientevent,flags=[COMPOUND],params=TableParams{comment=, read_repair_chance=0.0, dclocal_read_repair_chance=0.1, bloom_filter_fp_chance=0.01, crc_check_chance=1.0, gc_grace_seconds=864000, default_time_to_live=0, memtable_flush_period_in_ms=0, min_index_interval=128, max_index_interval=2048, speculative_retry=99PERCENTILE, caching={'keys' : 'ALL', 'rows_per_partition' : 'NONE'}, compaction=CompactionParams{class=org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy, options={min_threshold=4, max_threshold=32}}, compression=org.apache.cassandra.schema.CompressionParams@57a6e29, extensions={}, cdc=false},comparator=comparator(),partitionColumns=[[] | [client createdat name product]],partitionKeyColumns=[id],clusteringColumns=[],keyValidator=org.apache.cassandra.db.marshal.TimeUUIDType,columnMetadata=[product, id, client, name, createdat],droppedColumns={},triggers=[],indexes=[]]
2018-05-31 22:43:53.619 INFO 14976 --- [igrationStage:1] o.apache.cassandra.db.ColumnFamilyStore : Initializing testkeyspace.clientevent
我的配置中是否遗漏了什么？如何避免在测试之间等待一秒钟？
好的，我在 this post on CassandraUnit 之后做了一些更改，这不仅解决了我的问题，而且使我的测试 运行 快得多，因为它只创建一次数据库！
我对 EventsIntegrationTest 进行了以下更改：
@SpringBootTest(classes = EventsApplication.class, webEnvironment = RANDOM_PORT)
@AutoConfigureStubRunner(ids = ["example.com:clients-service",
"example.com:products-service"], stubsMode = StubsMode.LOCAL)
// no more TestExecutionListeners and cassandra annotations
@ActiveProfiles("test")
class EventsIntegrationTests extends Specification {
@Shared
private static Cluster cluster
@Shared
private static Session session
@Inject
StubTrigger stubTrigger
@Inject
private TestRestTemplate restTemplate
// runs only once before all tests, like JUnit 5's @BeforeAll
def setupSpec() {
// the same code from the post, creates the DB.
if (session == null) {
try {
EmbeddedCassandraServerHelper.startEmbeddedCassandra()
cluster = new Cluster.Builder().addContactPoint("localhost").withPort(9142).build()
session = cluster.connect()
CQLDataLoader loader = new CQLDataLoader(session)
ClassPathCQLDataSet dataSet = new ClassPathCQLDataSet("dataset.cql", true, true, "testkeyspace")
loader.load(dataSet)
} catch (Exception e) {
throw new RuntimeException("Could not start cassandra server or obtain a valid session.", e);
}
}
}
// same tests
// runs after every test like JUnit 5's @BeforeEach
def cleanup() {
// truncates the tables
Collection<TableMetadata> tables = cluster.getMetadata().getKeyspace("testkeyspace").getTables()
// Groovy's clojure, slightly different syntax from Java's Lambda
tables.forEach({session.execute(QueryBuilder.truncate(it))})
}
}
当然，因为我只有一个集成测试，所以所有代码都在那class。但是，如果您有多个涉及 Cassandra 的集成测试，我建议您使用这些方法创建一个基础 class。
我将 Cassandra 用作我的应用程序的事件存储，Spring Cloud Stream 接收事件并 Spring Cloud Cassandra 保存它们。
为了监听不同的事件，我创建了一个具有两种用户定义类型的 ClientEvent：客户端和产品（类 都用 @UserDefinedType 注释），如下所示：
@Table
public class ClientEvent {
@PrimaryKey
@CassandraType(type = DataType.Name.UUID)
private String id; // created with UUIDs.timeBased().toString()
@CassandraType(type = DataType.Name.UDT, userTypeName = "client")
private Client client; // has an email field
@CassandraType(type = DataType.Name.UDT, userTypeName = "product")
private Product product; // has name and cost fields
}
我的 EventConsumer 看起来像这样，并且适用于不同的事件：
@EnableBinding(Sink.class)
public class EventConsumer {
@Inject
private EventRepository eventRepository;
@Override
@StreamListener(Sink.INPUT)
public void clientEvent(ClientEvent event) {
eventRepository.save(event);
}
}
我还有一个控制器 returns 给定客户端的所有事件（所述控制器的代码非常标准，所以我不会在这里 post 它）。
为了测试所有这些，我设置了一些集成测试 spring-cloud-contract-stub-运行ner 和 cassandra-unit-spring 到运行 嵌入式卡桑德拉。我只有以下配置，在我的 application.yml:
spring:
profiles: test
data:
cassandra:
keyspace-name: testkeyspace
contact-points: localhost
port: 9142
这是 Spock 测试：
@SpringBootTest(classes = EventsApplication.class, webEnvironment = RANDOM_PORT)
@AutoConfigureStubRunner(ids = ["example.com:clients-service",
"example.com:products-service"], stubsMode = StubsMode.LOCAL)
@TestExecutionListeners(listeners = [
CassandraUnitDependencyInjectionTestExecutionListener.class,
DependencyInjectionTestExecutionListener.class
])
@CassandraDataSet(keyspace = "testkeyspace", value = "dataset.cql")
@EmbeddedCassandra(timeout = 60000L)
@ActiveProfiles("test")
class EventsIntegrationTests extends Specification {
@Inject
StubTrigger stubTrigger
@Inject
private TestRestTemplate restTemplate
def "client created"() {
given: "that a client has been created"
stubTrigger.trigger("client_created") //stub defined in clients-service
when: "I fetch the events for the current client"
def response = restTemplate.getForEntity("/", ClientEvent[])
then: "it should be a clientCreated event"
ClientEvent event = response.body[0]
event.name == "clientCreated"
event.client == new Client("test") // hashCode and equals implemented
}
def "product bought"() {
given: "that a product has been bought"
stubTrigger.trigger("product_bought") //stub defined in products-service
}
when: "I fetch the events for the current client"
def response = restTemplate.getForEntity("/", ClientEvent[])
then: "it should be a productBought event"
ClientEvent event = response.body[0]
event.name == "productBought"
event.client == new Client("test") // hashCode and equals implemented
event.product == new Product("Lamp", 100D)
}
}
dataset.cql 文件创建测试密钥空间、客户端和产品类型，然后是 ClientEvent table。到目前为止，一切都很好。问题是，虽然第一个测试 运行 没问题，但第二个测试在 EventConsumer 中保存事件时给了我这个 NullPointerException:
org.springframework.messaging.MessagingException: Exception thrown while invoking com.example.events.consumers.EventConsumer#clientEvent[1 args]; nested exception is java.lang.NullPointerException
[...]
Caused by: java.lang.NullPointerException: null
at org.springframework.data.cassandra.core.mapping.SimpleUserTypeResolver.resolveType(SimpleUserTypeResolver.java:63) ~[spring-data-cassandra-2.0.6.RELEASE.jar:2.0.6.RELEASE]
我发现如果我在每次测试之间调试或等待一秒钟（在 setup() 方法中使用 Thread.sleep(1000)），我的测试 运行 很好！我注意到在每次测试之间，cassandra-unit-spring 删除并重新创建数据库。在我看来，第二次测试发生在重新创建用户定义类型之前，这就是它抛出 NullPointerException 的原因：
2018-05-31 22:43:43.552 INFO 14976 --- [port-Requests-1] o.a.cassandra.service.MigrationManager : Drop Keyspace 'testkeyspace'
2018-05-31 22:43:53.445 INFO 14976 --- [igrationStage:1] o.a.cassandra.utils.memory.BufferPool : Global buffer pool is enabled, when pool is exhausted (max is 512.000MiB) it will allocate on heap
2018-05-31 22:43:53.511 INFO 14976 --- [port-Requests-3] o.a.cassandra.service.MigrationManager : Create new Keyspace: KeyspaceMetadata{name=testkeyspace, params=KeyspaceParams{durable_writes=false, replication=ReplicationParams{class=org.apache.cassandra.locator.SimpleStrategy, replication_factor=1}}, tables=[], views=[], functions=[], types=[]}
2018-05-31 22:43:53.603 INFO 14976 --- [port-Requests-2] o.a.cassandra.service.MigrationManager : Create new table: org.apache.cassandra.config.CFMetaData@1f5471ca[cfId=3d9ed530-653d-11e8-b838-27af117b0453,ksName=testkeyspace,cfName=clientevent,flags=[COMPOUND],params=TableParams{comment=, read_repair_chance=0.0, dclocal_read_repair_chance=0.1, bloom_filter_fp_chance=0.01, crc_check_chance=1.0, gc_grace_seconds=864000, default_time_to_live=0, memtable_flush_period_in_ms=0, min_index_interval=128, max_index_interval=2048, speculative_retry=99PERCENTILE, caching={'keys' : 'ALL', 'rows_per_partition' : 'NONE'}, compaction=CompactionParams{class=org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy, options={min_threshold=4, max_threshold=32}}, compression=org.apache.cassandra.schema.CompressionParams@57a6e29, extensions={}, cdc=false},comparator=comparator(),partitionColumns=[[] | [client createdat name product]],partitionKeyColumns=[id],clusteringColumns=[],keyValidator=org.apache.cassandra.db.marshal.TimeUUIDType,columnMetadata=[product, id, client, name, createdat],droppedColumns={},triggers=[],indexes=[]]
2018-05-31 22:43:53.619 INFO 14976 --- [igrationStage:1] o.apache.cassandra.db.ColumnFamilyStore : Initializing testkeyspace.clientevent
我的配置中是否遗漏了什么？如何避免在测试之间等待一秒钟？
好的，我在 this post on CassandraUnit 之后做了一些更改，这不仅解决了我的问题，而且使我的测试 运行 快得多，因为它只创建一次数据库！
我对 EventsIntegrationTest 进行了以下更改：
@SpringBootTest(classes = EventsApplication.class, webEnvironment = RANDOM_PORT)
@AutoConfigureStubRunner(ids = ["example.com:clients-service",
"example.com:products-service"], stubsMode = StubsMode.LOCAL)
// no more TestExecutionListeners and cassandra annotations
@ActiveProfiles("test")
class EventsIntegrationTests extends Specification {
@Shared
private static Cluster cluster
@Shared
private static Session session
@Inject
StubTrigger stubTrigger
@Inject
private TestRestTemplate restTemplate
// runs only once before all tests, like JUnit 5's @BeforeAll
def setupSpec() {
// the same code from the post, creates the DB.
if (session == null) {
try {
EmbeddedCassandraServerHelper.startEmbeddedCassandra()
cluster = new Cluster.Builder().addContactPoint("localhost").withPort(9142).build()
session = cluster.connect()
CQLDataLoader loader = new CQLDataLoader(session)
ClassPathCQLDataSet dataSet = new ClassPathCQLDataSet("dataset.cql", true, true, "testkeyspace")
loader.load(dataSet)
} catch (Exception e) {
throw new RuntimeException("Could not start cassandra server or obtain a valid session.", e);
}
}
}
// same tests
// runs after every test like JUnit 5's @BeforeEach
def cleanup() {
// truncates the tables
Collection<TableMetadata> tables = cluster.getMetadata().getKeyspace("testkeyspace").getTables()
// Groovy's clojure, slightly different syntax from Java's Lambda
tables.forEach({session.execute(QueryBuilder.truncate(it))})
}
}
当然，因为我只有一个集成测试，所以所有代码都在那class。但是，如果您有多个涉及 Cassandra 的集成测试，我建议您使用这些方法创建一个基础 class。