clean up sources after dropping streaming jobs.

src/meta/src/barrier/schedule.rs

@@ -229,7 +229,7 @@ impl BarrierScheduler {
         }
     }
 
-    /// Run multiple commands and return when they're all completely finished. It's ensured that
+    /// Run multiple commands and return when they're all completely finished (i.e., collected). It's ensured that
     /// multiple commands are executed continuously.
     ///
     /// Returns the barrier info of each command.
@@ -298,7 +298,7 @@ impl BarrierScheduler {
         .await
     }
 
-    /// Run a command and return when it's completely finished.
+    /// Run a command and return when it's completely finished (i.e., collected).
     ///
     /// Returns the barrier info of the actual command.
     pub async fn run_command(&self, database_id: DatabaseId, command: Command) -> MetaResult<()> {

src/meta/src/rpc/ddl_controller.rs

@@ -1256,13 +1256,23 @@ impl DdlController {
             ..
         } = release_ctx;
 
-        // unregister sources.
+        self.stream_manager
+            .drop_streaming_jobs(
+                risingwave_common::catalog::DatabaseId::new(database_id as _),
+                removed_actors.into_iter().map(|id| id as _).collect(),
+                removed_streaming_job_ids,
+                removed_state_table_ids,
+                removed_fragments.iter().map(|id| *id as _).collect(),
+            )
+            .await;
+
+        // clean up sources after dropping streaming jobs.
+        // Otherwise, e.g., Kafka consumer groups might be recreated after deleted.
         self.source_manager
             .apply_source_change(SourceChange::DropSource {
                 dropped_source_ids: removed_source_ids.into_iter().map(|id| id as _).collect(),
             })
             .await;
-
         // unregister fragments and actors from source manager.
         // FIXME: need also unregister source backfill fragments.
         let dropped_source_fragments = removed_source_fragments
@@ -1282,17 +1292,6 @@ impl DdlController {
             })
             .await;
 
-        // drop streaming jobs.
-        self.stream_manager
-            .drop_streaming_jobs(
-                risingwave_common::catalog::DatabaseId::new(database_id as _),
-                removed_actors.into_iter().map(|id| id as _).collect(),
-                removed_streaming_job_ids,
-                removed_state_table_ids,
-                removed_fragments.iter().map(|id| *id as _).collect(),
-            )
-            .await;
-
         Ok(version)
     }