jprdonnelly/airbyte
1
0
Fork 0
mirror of synced 2025-12-25 02:09:19 -05:00
Code Issues Packages Projects Releases Wiki Activity

Track stream status in source (#24971)

Browse Source 
* WIP Track stream status in source

* Revert formatting

* Revert formatting changes

* Remove unnecessary file

* Automated Change

* Automated Change

* Use new stream status trace message

* Rename class

* Remove unnecessary import

* Formatting

* Add tests

* Fix compile issues

* Automated Commit - Formatting Changes

* Remove TODO

* Fix compilation error

* Split STOPPED into INCOMPLETE and COMPLETE

* Remove unused import

* Changelog updates for source-postgres

* Remove unused import

* auto-bump connector version

---------

Co-authored-by: jdpgrailsdev <jdpgrailsdev@users.noreply.github.com>
Co-authored-by: Octavia Squidington III <octavia-squidington-iii@users.noreply.github.com>
This commit is contained in:
Jonathan Pearlin
2023-04-26 16:14:25 -04:00
committed by GitHub
parent 6b328704aa
commit a38af089e9
38 changed files with 592 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlSource.java
View File

@@ -26,6 +26,7 @@ import io.airbyte.commons.util.AutoCloseableIterators;
import io.airbyte.db.factory.DatabaseDriver;
import io.airbyte.db.jdbc.JdbcDatabase;
import io.airbyte.db.jdbc.JdbcUtils;
import io.airbyte.integrations.base.AirbyteTraceMessageUtility;
import io.airbyte.integrations.base.IntegrationRunner;
import io.airbyte.integrations.base.Source;
import io.airbyte.integrations.base.ssh.SshWrappedSource;
@@ -292,7 +293,8 @@ public class MySqlSource extends AbstractJdbcSource<MysqlType> implements Source
emittedAt);
return Collections.singletonList(
AutoCloseableIterators.concatWithEagerClose(snapshotIterator, AutoCloseableIterators.lazyIterator(incrementalIteratorSupplier)));
AutoCloseableIterators.concatWithEagerClose(AirbyteTraceMessageUtility::emitStreamStatusTrace, snapshotIterator,
AutoCloseableIterators.lazyIterator(incrementalIteratorSupplier, null)));
} else {
LOGGER.info("using CDC: {}", false);
return super.getIncrementalIterators(database, catalog, tableNameToTable, stateManager,