Try to fix add new table crash

[yatsvic]

We will have overhead. If database.history.store.only.monitored.tables.ddl is set to true on each connector start it will additionally load tables list from database.

p.s. store.only.captured.tables.ddl is deprecated, should use store.only.captured.tables.ddl instead

[wadesherman]

this is great - i will test this locally

[wadesherman]

do you think it is feasible to just call createSchemaEventsForTables right here?

[yatsvic]

@wadesherman I have tried

    private void ensureSchemaForNewTables(MySqlPartition partition) {
        try (MySqlSnapshotContext snapshotContext = (MySqlSnapshotContext) prepare(partition)) {
            determineCapturedTables(snapshotContext);
            boolean newTables = !databaseSchema.tableIds().containsAll(snapshotContext.capturedSchemaTables);
            LOGGER.info("new tables detected: " + newTables);
            if (newTables) {
                ChangeEventSourceContext sourceContext = new ChangeEventSourceContext() {

                    @Override
                    public boolean isRunning() {
                        return true;
                    }
                };

                SnapshottingTask task = new SnapshottingTask(true, false);
                createSchemaChangeEventsForTables(sourceContext, snapshotContext, task);
            }

        }
        catch (Exception e) {
            LOGGER.error("Failed to recapture new tables schemas.", e);
            throw new RuntimeException(e);
        }
    }

    @Override
    protected SnapshottingTask getSnapshottingTask(MySqlPartition partition, MySqlOffsetContext previousOffset) {

        boolean snapshotSchema = true;
        boolean snapshotData = true;

        // found a previous offset and the earlier snapshot has completed
        if (previousOffset != null && !previousOffset.isSnapshotRunning()) {
            LOGGER.info("A previous offset indicating a completed snapshot has been found. Neither schema nor data will be snapshotted.");
            snapshotSchema = databaseSchema.isStorageInitializationExecuted();
            if (!snapshotSchema && databaseSchema.storeOnlyCapturedTables()) {
                ensureSchemaForNewTables(partition);
            }
            snapshotData = false;
        }
...

But it doesn't produce schema messages. Looks like that something is not inititialized. I think that solution in this branch is good. The only one thing is memory overhead: it recapture all tables schemas. But it uses standard snapshot procedure with locks and interruption on connector stop.