DebeziumServer { DebeziumEngine.ChangeConsumer<?> getConsumer() { return consumer; }  @SuppressWarnings("unchecked") @PostConstruct void start(); void stop(@Observes ShutdownEvent event);  }
@Test public void testJson() throws Exception { final TestConsumer testConsumer = (TestConsumer) server.getConsumer(); Awaitility.await().atMost(Duration.ofSeconds(TestConfigSource.waitForSeconds())).until(() -> (testConsumer.getValues().size() >= MESSAGE_COUNT)); Assertions.assertThat(testConsumer.getValues().size()).isEqualTo(MESSAGE_COUNT); Assertions.assertThat(testConsumer.getValues().get(MESSAGE_COUNT - 1)).isEqualTo("{\"line\":\"" + MESSAGE_COUNT + "\"}"); }