private static void ensureBuildVersionTable(final Configuration txnConfig,
                                                final Connection connection,
                                                final String thisNodeName) {
        try {
            // Try to create it in case this is an old version without it
            createBuildVersionTableIfNotExists(txnConfig);

            // Now check whatever flavour of the table is there has the right cols
            final String listColsSql = """
                    SELECT column_name
                    FROM information_schema.columns
                    WHERE table_schema = schema()
                    AND table_name = ?""";

            final Set<Object> columnNames = DSL.using(txnConfig)
                    .fetch(listColsSql, BUILD_VERSION_TABLE_NAME)
                    .stream()
                    .map(record -> record.get(0))
                    .collect(Collectors.toSet());

            if (!columnNames.containsAll(ALL_COLUMNS)) {
                // This is an old version of the table, so we need to re-create it with the right cols

                // Another node may have already dropped it so allow for it to not exist
                final String dropTableSql = LogUtil.message("""
                                DROP TABLE IF EXISTS {}""",
                        BUILD_VERSION_TABLE_NAME);
                // Causes implicit commit
                DSL.using(txnConfig)
                        .execute(dropTableSql);

                // Another node my beat us to this
                createBuildVersionTableIfNotExists(txnConfig);
            }

            // Do a select first to avoid being blocked by the row lock with the insert stmt below
            final boolean isRecordPresent = DSL.using(txnConfig)
                    .fetchOptional(LogUtil.message("""
                                            SELECT {}
                                            FROM {}
                                            WHERE id = ?""",
                                    BUILD_VERSION_COL_NAME,
                                    BUILD_VERSION_TABLE_NAME),
                            BUILD_VERSION_TABLE_ROW_ID)
                    .isPresent();

            if (!isRecordPresent) {
                // Ensure we have a record that we can get a lock on
                // Done like this in case another node gets in before us.
                // This may block if another node beat us to it and locked the row but
                // that is fine.
                final String insertSql = LogUtil.message("""
                                INSERT INTO {} (
                                    {},
                                    {},
                                    {})
                                SELECT ?, ?, ?
                                FROM DUAL
                                WHERE NOT EXISTS (
                                    SELECT NULL
                                    FROM {}
                                    WHERE ID = ?)
                                LIMIT 1""",
                        BUILD_VERSION_TABLE_NAME,
                        ID_COL_NAME,
                        BUILD_VERSION_COL_NAME,
                        EXECUTING_NODE_COL_NAME,
                        BUILD_VERSION_TABLE_NAME);

                final int result = DSL.using(txnConfig)
                        .execute(insertSql,
                                BUILD_VERSION_TABLE_ROW_ID,
                                INITIAL_BUILD_VERSION,
                                thisNodeName,
                                BUILD_VERSION_TABLE_ROW_ID);
                // Make sure other nodes can see it
                if (result > 0) {
                    LOGGER.info("Committing new {} row", BUILD_VERSION_TABLE_NAME);
                    connection.commit();
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("Error ensuring table "
                                       + BUILD_VERSION_TABLE_NAME + ": "
                                       + e.getMessage(), e);
        }
    }
Use a table lock on a noddy table created just for this purpose to enforce
a cluster wide lock so we can do all the flyway migrations in isolation.
Flyway migrations should work under lock but we have lots of independent flyway
migrations so it is safer to do it this way.
Can't user the cluster lock service as that is not a thing at this point.
stroom-app/src/main/java/stroom/app/BootstrapUtil.java