[test][e2e] Clean up database resources for every e2e tests.

flink-cdc-e2e-tests/src/test/java/com/ververica/cdc/connectors/tests/utils/FlinkContainerTestEnvironment.java

@@ -165,6 +165,7 @@ public abstract class FlinkContainerTestEnvironment extends TestLogger {
         if (taskManager != null) {
             taskManager.stop();
         }
+        mysqlInventoryDatabase.dropDatabase();
     }
 
     /**

flink-connector-mysql-cdc/src/test/java/com/ververica/cdc/connectors/mysql/testutils/UniqueDatabase.java

@@ -48,6 +48,7 @@ public class UniqueDatabase {
 
     private static final String[] CREATE_DATABASE_DDL =
             new String[] {"CREATE DATABASE $DBNAME$;", "USE $DBNAME$;"};
+    private static final String DROP_DATABASE_DDL = "DROP DATABASE IF EXISTS $DBNAME$;";
     private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");
 
     private final MySqlContainer container;
@@ -143,6 +144,21 @@ public class UniqueDatabase {
         }
     }
 
+    /** Drop the database if is exists. */
+    public void dropDatabase() {
+        try {
+            try (Connection connection =
+                            DriverManager.getConnection(
+                                    container.getJdbcUrl(), username, password);
+                    Statement statement = connection.createStatement()) {
+                final String dropDatabaseStatement = convertSQL(DROP_DATABASE_DDL);
+                statement.execute(dropDatabaseStatement);
+            }
+        } catch (final Exception e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
     public Connection getJdbcConnection() throws SQLException {
         return DriverManager.getConnection(container.getJdbcUrl(databaseName), username, password);
     }