0

我使用 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);
    }
}

我还有一个控制器，它返回给定客户端的所有事件（所述控制器的代码非常标准，所以我不会在这里发布）。

为了测试这一切，我使用 spring-cloud-contract-stub-runner 和 cassandra-unit-spring 设置了一些集成测试来运行嵌入式 cassandra。我的 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 文件创建 testkeyspace、客户端和产品类型，然后创建 ClientEvent 表。到目前为止，一切都很好。问题是，虽然第一个测试运行良好，但在 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]

我发现如果我在每次测试之间调试或等待一秒钟（使用Thread.sleep(1000)方法setup()），我的测试运行良好！我注意到在每次测试之间，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

我的配置中是否缺少某些内容？如何避免在测试之间等待一秒钟？

4

1 回答 1

0

好的，我在 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))})
    }
}

当然，由于我只有一个集成测试，所以所有代码都在那个类中。但是，如果您有多个涉及 Cassandra 的集成测试，我建议您使用这些方法创建一个基类。

于 2018-06-07T13:03:08.570 回答