Properly implement replacement for Flow<T>.concatWith(flow) to avoid deadlocks when the downstream throws, but other flow has no elements in it (Kotlin#2659)

Fixes KT-46013

Author: qwwdfsad

kotlinx-coroutines-core/common/src/flow/Migration.kt

@@ -344,13 +344,13 @@ public fun <T> Flow<T>.concatWith(value: T): Flow<T> = noImpl()
 
 /**
  * Flow analogue of `concatWith` is [onCompletion].
- * Use `onCompletion { emitAll(other) }`.
+ * Use `onCompletion { if (it == null) emitAll(other) }`.
  * @suppress
  */
 @Deprecated(
     level = DeprecationLevel.ERROR,
-    message = "Flow analogue of 'concatWith' is 'onCompletion'. Use 'onCompletion { emitAll(other) }'",
-    replaceWith = ReplaceWith("onCompletion { emitAll(other) }")
+    message = "Flow analogue of 'concatWith' is 'onCompletion'. Use 'onCompletion { if (it == null) emitAll(other) }'",
+    replaceWith = ReplaceWith("onCompletion { if (it == null) emitAll(other) }")
 )
 public fun <T> Flow<T>.concatWith(other: Flow<T>): Flow<T> = noImpl()
 
@@ -404,7 +404,7 @@ public fun <T1, T2, T3, T4, T5, R> Flow<T1>.combineLatest(
  * @suppress
  */
 @Deprecated(
-    level = DeprecationLevel.WARNING, // since 1.3.0, error in 1.4.0
+    level = DeprecationLevel.ERROR, // since 1.3.0, error in 1.5.0
     message = "Use 'onStart { delay(timeMillis) }'",
     replaceWith = ReplaceWith("onStart { delay(timeMillis) }")
 )
@@ -416,7 +416,7 @@ public fun <T> Flow<T>.delayFlow(timeMillis: Long): Flow<T> = onStart { delay(ti
  * @suppress
  */
 @Deprecated(
-    level = DeprecationLevel.WARNING, // since 1.3.0, error in 1.4.0
+    level = DeprecationLevel.ERROR, // since 1.3.0, error in 1.5.0
     message = "Use 'onEach { delay(timeMillis) }'",
     replaceWith = ReplaceWith("onEach { delay(timeMillis) }")
 )
@@ -430,7 +430,7 @@ public fun <T> Flow<T>.delayEach(timeMillis: Long): Flow<T> = onEach { delay(tim
 public fun <T, R> Flow<T>.switchMap(transform: suspend (value: T) -> Flow<R>): Flow<R> = flatMapLatest(transform)
 
 @Deprecated(
-    level = DeprecationLevel.WARNING, // Since 1.3.8, was experimental when deprecated
+    level = DeprecationLevel.ERROR, // Warning since 1.3.8, was experimental when deprecated, ERROR since 1.5.0
     message = "'scanReduce' was renamed to 'runningReduce' to be consistent with Kotlin standard library",
     replaceWith = ReplaceWith("runningReduce(operation)")
 )