alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-518 Remove a leftover event

Browse Source
This commit is contained in:
Jiri Pechanec 2017-12-19 14:09:43 +01:00
parent ecf6a86594
commit 42d103294e
1 changed files with 24 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
debezium-connector-postgres/src/test/java/io/debezium/connector/postgresql/RecordsStreamProducerIT.java
View File

@ -385,6 +385,22 @@ public void shouldProcessIntervalDelete() throws Exception {
consumer = testConsumer(4);
recordsProducer.start(consumer);
executeAndWait(statements);
final String topicPrefix = "public.table_with_interval";
final String topicName = topicName(topicPrefix);
final String pk = "id";
assertRecordInserted(topicPrefix, pk, 1);
assertRecordInserted(topicPrefix, pk, 2);
// first entry removed
SourceRecord record = consumer.remove();
assertEquals(topicName, record.topic());
VerifyRecord.isValidDelete(record, pk, 1);
// followed by a tombstone
record = consumer.remove();
assertEquals(topicName, record.topic());
VerifyRecord.isValidTombstone(record, pk, 1);
}
@Test
@ -392,6 +408,14 @@ public void shouldProcessIntervalDelete() throws Exception {
public void shouldNotStartAfterStop() throws Exception {
recordsProducer.stop();
recordsProducer.start(consumer);
// Need to remove record created in @Before
PostgresConnectorConfig config = new PostgresConnectorConfig(TestHelper.defaultConfig().with(PostgresConnectorConfig.INCLUDE_UNKNOWN_DATATYPES, true).build());
PostgresTaskContext context = new PostgresTaskContext(config, new PostgresSchema(config));
recordsProducer = new RecordsStreamProducer(context, new SourceInfo(config.serverName()));
consumer = testConsumer(1);
recordsProducer.start(consumer);
}
private void assertInsert(String statement, List<SchemaAndValueField> expectedSchemaAndValuesByColumn) {