KafkaDatabaseHistory extends AbstractDatabaseHistory { @Override public boolean exists() { boolean exists = false; if (storageExists()) { try (KafkaConsumer<String, String> historyConsumer = new KafkaConsumer<>(consumerConfig.asProperties());) { checkTopicSettings(topicName); Set<TopicPartition> historyTopic = Collections.singleton(new TopicPartition(topicName, PARTITION)); Map<TopicPartition, Long> beginningOffsets = historyConsumer.beginningOffsets(historyTopic); Map<TopicPartition, Long> endOffsets = historyConsumer.endOffsets(historyTopic); Long beginOffset = beginningOffsets.entrySet().iterator().next().getValue(); Long endOffset = endOffsets.entrySet().iterator().next().getValue(); exists = endOffset > beginOffset; } } return exists; }  @Override void configure(Configuration config, HistoryRecordComparator comparator, DatabaseHistoryListener listener, boolean useCatalogBeforeSchema); @Override synchronized void start(); @Override boolean storageExists(); @Override boolean exists(); @Override synchronized void stop(); @Override String toString(); @Override void initializeStorage(); static final Field TOPIC; static final Field BOOTSTRAP_SERVERS; static final Field RECOVERY_POLL_INTERVAL_MS; static final Field RECOVERY_POLL_ATTEMPTS; static final Field INTERNAL_CONNECTOR_CLASS; static final Field INTERNAL_CONNECTOR_ID; static Field.Set ALL_FIELDS; }
@Test public void testExists() { String topicName = "exists-schema-changes"; testHistoryTopicContent(topicName, true); assertTrue(history.exists()); Configuration config = Configuration.create() .with(KafkaDatabaseHistory.BOOTSTRAP_SERVERS, kafka.brokerList()) .with(KafkaDatabaseHistory.TOPIC, "dummytopic") .with(DatabaseHistory.NAME, "my-db-history") .with(KafkaDatabaseHistory.RECOVERY_POLL_INTERVAL_MS, 500) .with(KafkaDatabaseHistory.consumerConfigPropertyName( ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG), 100) .with(KafkaDatabaseHistory.consumerConfigPropertyName( ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG), 50000) .with(KafkaDatabaseHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS, true) .with(KafkaDatabaseHistory.INTERNAL_CONNECTOR_CLASS, "org.apache.kafka.connect.source.SourceConnector") .with(KafkaDatabaseHistory.INTERNAL_CONNECTOR_ID, "dbz-test") .build(); history.configure(config, null, DatabaseHistoryMetrics.NOOP, true); history.start(); assertFalse(history.exists()); }