fix: test fixes

src/main/java/io/supertokens/storage/mysql/BulkImportProxyStorage.java

@@ -44,7 +44,7 @@ public synchronized Connection getTransactionConnection() throws SQLException, S
         if (this.connection == null) {
             Connection con = ConnectionPool.getConnectionForProxyStorage(this);
             this.connection = new BulkImportProxyConnection(con);
-            connection.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
+            connection.setTransactionIsolation(Connection.TRANSACTION_SERIALIZABLE); // TODO: Change this to READ COMMITTED when mysql 5.7 support is dropped and SKIP LOCKED is implemented in loading the bulkimportUsers // https://github.com/supertokens/supertokens-mysql-plugin/issues/141
             connection.setAutoCommit(false);
         }
         return this.connection;

src/main/java/io/supertokens/storage/mysql/Start.java

@@ -3759,23 +3759,32 @@ public int getDbActivityCount(String dbname) throws SQLException, StorageQueryEx
 
     @Override
     public void addBulkImportUsers(AppIdentifier appIdentifier, List<BulkImportUser> users)
-            throws StorageQueryException,
-            TenantOrAppNotFoundException,
-            io.supertokens.pluginInterface.bulkimport.exceptions.DuplicateUserIdException {
+            throws StorageQueryException {
         try {
-            BulkImportQueries.insertBulkImportUsers(this, appIdentifier, users);
-        } catch (SQLException e) {
-            if (e instanceof SQLIntegrityConstraintViolationException) {
-                MySQLConfig config = Config.getConfig(this);
-                String errorMessage = e.getMessage();
-                if (isPrimaryKeyError(errorMessage, config.getBulkImportUsersTable())) {
-                    throw new io.supertokens.pluginInterface.bulkimport.exceptions.DuplicateUserIdException();
-                }
-                if (isForeignKeyConstraintError(errorMessage, config.getBulkImportUsersTable(), "app_id")) {
-                    throw new TenantOrAppNotFoundException(appIdentifier);
+            this.startTransaction(con -> {
+                try {
+                    BulkImportQueries.insertBulkImportUsers_Transaction(this, (Connection) con.getConnection(), appIdentifier, users);
+                } catch (SQLException e) {
+                    if (e instanceof SQLIntegrityConstraintViolationException) {
+                        MySQLConfig config = Config.getConfig(this);
+                        String errorMessage = e.getMessage();
+                        if (isPrimaryKeyError(errorMessage, config.getBulkImportUsersTable())) {
+                            throw new StorageTransactionLogicException(new io.supertokens.pluginInterface.bulkimport.exceptions.DuplicateUserIdException());
+                        }
+                        if (isForeignKeyConstraintError(errorMessage, config.getBulkImportUsersTable(), "app_id")) {
+                            throw new TenantOrAppNotFoundException(appIdentifier);
+                        }
+                    }
+                    throw new StorageQueryException(e);
                 }
+                return null;
+            });
+        } catch (StorageTransactionLogicException e) {
+            if(e.actualException instanceof StorageQueryException) {
+                throw (StorageQueryException) e.actualException;
+            } else {
+                throw new StorageQueryException(e.actualException);
             }
-            throw new StorageQueryException(e);
         }
     }
 

src/main/java/io/supertokens/storage/mysql/queries/BulkImportQueries.java

@@ -71,31 +71,24 @@ public static String getQueryToCreatePaginationIndex2(Start start) {
                 + Config.getConfig(start).getBulkImportUsersTable() + " (app_id, created_at DESC, id DESC)";
     }
 
-    public static void insertBulkImportUsers(Start start, AppIdentifier appIdentifier, List<BulkImportUser> users)
+    public static void insertBulkImportUsers_Transaction(Start start, Connection con, AppIdentifier appIdentifier, List<BulkImportUser> users)
             throws SQLException, StorageQueryException {
-        StringBuilder queryBuilder = new StringBuilder(
-                "INSERT INTO " + Config.getConfig(start).getBulkImportUsersTable() + " (id, app_id, raw_data, created_at, updated_at) VALUES ");
+        String queryBuilder = "INSERT INTO " + Config.getConfig(start).getBulkImportUsersTable() +
+                " (id, app_id, raw_data, created_at, updated_at) VALUES "
+                + " (?, ?, ?, ?, ?)";
 
-        int userCount = users.size();
-
-        for (int i = 0; i < userCount; i++) {
-            queryBuilder.append(" (?, ?, ?, ?, ?)");
-
-            if (i < userCount - 1) {
-                queryBuilder.append(",");
-            }
-        }
-
-        update(start, queryBuilder.toString(), pst -> {
-            int parameterIndex = 1;
+        List<PreparedStatementValueSetter> valueSetters = new ArrayList<>();
             for (BulkImportUser user : users) {
-                pst.setString(parameterIndex++, user.id);
-                pst.setString(parameterIndex++, appIdentifier.getAppId());
-                pst.setString(parameterIndex++, user.toRawDataForDbStorage());
-                pst.setLong(parameterIndex++, System.currentTimeMillis());
-                pst.setLong(parameterIndex++, System.currentTimeMillis());
+                valueSetters.add(pst -> {
+                    pst.setString(1, user.id);
+                    pst.setString(2, appIdentifier.getAppId());
+                    pst.setString(3, user.toRawDataForDbStorage());
+                    pst.setLong(4, System.currentTimeMillis());
+                    pst.setLong(5, System.currentTimeMillis());
+                });
             }
-        });
+
+        executeBatch(con, queryBuilder, valueSetters);
     }
 
     public static void updateBulkImportUserStatus_Transaction(Start start, Connection con, AppIdentifier appIdentifier,
@@ -134,8 +127,8 @@ public static List<BulkImportUser> getBulkImportUsersAndChangeStatusToProcessing
                 // "SKIP LOCKED" allows other processes to skip locked rows and select the next 1000 available rows.
                 String selectQuery = "SELECT * FROM " + Config.getConfig(start).getBulkImportUsersTable()
                 + " WHERE app_id = ?"
-                + " AND (status = 'NEW' OR status = 'PROCESSING')" /* 10 mins */
-                + " LIMIT ? FOR UPDATE";
+                + " AND (status = 'NEW' OR status = 'PROCESSING')"
+                + " LIMIT ? FOR UPDATE"; // TODO: add 'SKIP LOCKED' when mysql 5.7 support is dropped // https://github.com/supertokens/supertokens-mysql-plugin/issues/141
 
 
                 List<BulkImportUser> bulkImportUsers = new ArrayList<>();
@@ -155,18 +148,20 @@ public static List<BulkImportUser> getBulkImportUsersAndChangeStatusToProcessing
                 }
 
                 String updateQuery = "UPDATE " + Config.getConfig(start).getBulkImportUsersTable()
-                        + " SET status = ?, updated_at = ? WHERE app_id = ? AND id IN (" + Utils
-                                .generateCommaSeperatedQuestionMarks(bulkImportUsers.size()) + ")";
-
-                update(sqlCon, updateQuery, pst -> {
-                    int index = 1;
-                    pst.setString(index++, BULK_IMPORT_USER_STATUS.PROCESSING.toString());
-                    pst.setLong(index++, System.currentTimeMillis());
-                    pst.setString(index++, appIdentifier.getAppId());
-                    for (BulkImportUser user : bulkImportUsers) {
-                        pst.setObject(index++, user.id);
-                    }
-                });
+                        + " SET status = ?, updated_at = ? WHERE app_id = ? AND id = ?";
+
+                List<PreparedStatementValueSetter> updateSetters = new ArrayList<>();
+                for(BulkImportUser user : bulkImportUsers){
+                    updateSetters.add(pst -> {
+                        pst.setString(1, BULK_IMPORT_USER_STATUS.PROCESSING.toString());
+                        pst.setLong(2, System.currentTimeMillis());
+                        pst.setString(3, appIdentifier.getAppId());
+                        pst.setObject(4, user.id);
+                    });
+                }
+
+                executeBatch(sqlCon, updateQuery, updateSetters);
+
                 return bulkImportUsers;
             } catch (SQLException throwables) {
                 throw new StorageTransactionLogicException(throwables);

src/test/java/io/supertokens/storage/mysql/test/OneMillionUsersTest.java

@@ -991,9 +991,9 @@ public void testWithOneMillionUsers() throws Exception {
             return;
         }
 
-        Main main = startCronProcess(String.valueOf(16));
+        Main main = startCronProcess(String.valueOf(4));
 
-        int NUMBER_OF_USERS_TO_UPLOAD = 500000; // half million
+        int NUMBER_OF_USERS_TO_UPLOAD = 100000; // hundred thousand users // change this when mysql 5.7 support is dropped // https://github.com/supertokens/supertokens-mysql-plugin/issues/141
 
         if (StorageLayer.getBaseStorage(main).getType() != STORAGE_TYPE.SQL || StorageLayer.isInMemDb(main)) {
             return;