OXY-146: Pulsar createIfDNE idempotency (Topic already exists)

build.sbt

@@ -389,6 +389,7 @@ lazy val `oxygen-events-pulsar`: Project =
     .dependsOn(
       `oxygen-events` % testAndCompile,
       `oxygen-schema`.jvm % testAndCompile,
+      `oxygen-test`.jvm % Test,
     )
 
 lazy val `oxygen-storage-in-memory`: Project =

modules/events/pulsar/src/main/scala/oxygen/pulsar/client/PulsarAdminClient.scala

@@ -1,6 +1,8 @@
 package oxygen.pulsar.client
 
+import java.util.concurrent.{CompletionException, ExecutionException}
 import org.apache.pulsar.client.admin.PulsarAdmin as RawPulsarAdminClient
+import org.apache.pulsar.client.admin.PulsarAdminException
 import org.apache.pulsar.common.policies.data.TenantInfo
 import oxygen.predef.core.*
 import oxygen.pulsar.model.*
@@ -49,7 +51,7 @@ final class PulsarAdminClient(client: RawPulsarAdminClient) { self =>
     def createIfDNE(tenant: PulsarTenant): Task[Unit] =
       list.flatMap {
         case tenants if tenants.contains(tenant) => ZIO.logInfo(s"Pulsar tenant already exists [${tenant.tenant}]")
-        case _                                   => create(tenant)
+        case _                                   => succeedIfAlreadyExists(s"Pulsar tenant already exists [${tenant.tenant}]")(create(tenant))
       }
 
   }
@@ -76,7 +78,7 @@ final class PulsarAdminClient(client: RawPulsarAdminClient) { self =>
       self.tenant.createIfDNE(tenant) *>
         list(tenant).flatMap {
           case namespaces if namespaces.contains(namespace) => ZIO.logInfo(s"Pulsar namespace already exists [${namespace.fullyQualified}]")
-          case _                                            => create(namespace)
+          case _                                            => succeedIfAlreadyExists(s"Pulsar namespace already exists [${namespace.fullyQualified}]")(create(namespace))
         }
     }
 
@@ -128,14 +130,45 @@ final class PulsarAdminClient(client: RawPulsarAdminClient) { self =>
         filtered = topics.filter(_.ignorePartition == baseTopic)
         numPartitions = filtered.flatMap(_.partition).length
         _ <-
-          if filtered.isEmpty then create(topic, partitions)
+          if filtered.isEmpty then succeedIfAlreadyExists(s"Pulsar topic already exists [${topic.fullyQualified}]")(create(topic, partitions))
           else if numPartitions != partitions.getOrElse(0) then ZIO.logWarning(s"Pulsar topic exists [${topic.fullyQualified}], but has mismatching partitions")
           else ZIO.logInfo(s"Pulsar topic already exists [${topic.fullyQualified}]")
       } yield ()
     }
 
   }
 
+  /**
+    * Runs a `create*` effect, treating an "already exists" failure as success.
+    *
+    * `createIfDNE` uses a check-then-act (list, then create) pattern which is not atomic: under
+    * concurrent calls two fibers can both observe the resource as missing and both attempt to
+    * create it, and the loser receives a 409 `PulsarAdminException` ("... already exists"). Since
+    * the post-condition (the resource exists) still holds, this is treated as success.
+    */
+  private def succeedIfAlreadyExists(alreadyExistsMessage: String)(create: Task[Unit]): Task[Unit] =
+    create.catchSome {
+      case error if isAlreadyExists(error) => ZIO.logInfo(alreadyExistsMessage)
+    }
+
+  /**
+    * Whether `error` represents an "already exists" / 409 conflict from a Pulsar `create*` call.
+    *
+    *   - A `PulsarAdminException` with HTTP status 409 is treated as already-exists.
+    *   - `CompletionException` / `ExecutionException` (as surfaced by `ZIO.fromCompletableFuture`)
+    *     are unwrapped and re-checked.
+    *   - Otherwise fall back to a case-insensitive check for "already exists" in the message.
+    */
+  private def isAlreadyExists(error: Throwable): Boolean =
+    error match {
+      case e: PulsarAdminException if e.getStatusCode == 409 =>
+        true
+      case (_: CompletionException | _: ExecutionException) if error.getCause != null && (error.getCause ne error) =>
+        isAlreadyExists(error.getCause)
+      case _ =>
+        Option(error.getMessage).exists(_.toLowerCase.contains("already exists"))
+    }
+
   private def listFlat[A, B](a: Task[ArraySeq[A]], b: A => Task[ArraySeq[B]]): Task[ArraySeq[B]] =
     a.flatMap(ZIO.foreach(_)(b)).map(_.flatten)
 

modules/events/pulsar/src/test/scala/oxygen/pulsar/client/PulsarAdminClientSpec.scala

@@ -0,0 +1,130 @@
+package oxygen.pulsar.client
+
+import java.lang.reflect.{InvocationHandler, Method, Proxy}
+import java.util.concurrent.{CompletableFuture, CompletionException}
+import org.apache.pulsar.client.admin.{Namespaces, PulsarAdmin, PulsarAdminException, Tenants, Topics}
+import oxygen.predef.test.*
+import oxygen.pulsar.model.*
+import scala.jdk.CollectionConverters.*
+
+/**
+  * Unit tests for [[PulsarAdminClient]] idempotency (OXY-146).
+  *
+  * `RawPulsarAdminClient` (`org.apache.pulsar.client.admin.PulsarAdmin`) is a large Java interface,
+  * so it is faked with a `java.lang.reflect.Proxy` that dispatches by method name. The tests drive
+  * the public `topic.createIfDNE`, arranging for the tenant + namespace to already exist and the
+  * topic to be missing so that a `create` is always attempted, then vary the failure the fake
+  * `create` returns.
+  */
+object PulsarAdminClientSpec extends OxygenSpecDefault {
+
+  private val tenant: PulsarTenant = PulsarTenant("test-tenant")
+  private val namespace: PulsarNamespace = PulsarNamespace("test-tenant", "test-ns")
+  private val topic: PulsarTopic = PulsarTopic(persistent = true, "test-tenant", "test-ns", "test-topic", None)
+
+  private def proxy[A](interface: Class[A])(handle: PartialFunction[(String, List[AnyRef]), AnyRef]): A = {
+    val invocationHandler: InvocationHandler =
+      (proxyInstance: Any, method: Method, args: Array[AnyRef]) => {
+        val argList: List[AnyRef] = Option(args).fold(List.empty[AnyRef])(_.toList)
+        method.getName match {
+          case "toString" => s"${interface.getSimpleName}$$mock"
+          case "hashCode" => Int.box(java.lang.System.identityHashCode(proxyInstance))
+          case "equals"   => Boolean.box(proxyInstance.asInstanceOf[AnyRef] eq argList.headOption.orNull)
+          case name       =>
+            handle.lift((name, argList)).getOrElse(throw new UnsupportedOperationException(s"${interface.getSimpleName}.$name"))
+        }
+      }
+    interface.cast(Proxy.newProxyInstance(interface.getClassLoader, Array[Class[?]](interface), invocationHandler))
+  }
+
+  private def completed[A](value: A): CompletableFuture[A] = CompletableFuture.completedFuture(value)
+
+  private def failed[A](error: Throwable): CompletableFuture[A] = {
+    val cf = new CompletableFuture[A]()
+    cf.completeExceptionally(error)
+    cf
+  }
+
+  private def conflict409: PulsarAdminException =
+    new PulsarAdminException.ConflictException(new RuntimeException("Topic already exists"), "Conflict", 409)
+
+  /**
+    * Builds a fake admin client where the tenant + namespace already exist. `topicList` controls
+    * what the topic listing returns, and `onCreate` supplies the result of the (non-partitioned)
+    * `create` call.
+    */
+  private def mockAdmin(
+      topicList: List[String],
+      onCreate: () => CompletableFuture[Void],
+  ): PulsarAdmin = {
+    val topics: Topics = proxy(classOf[Topics]) {
+      case ("getListAsync", _)                   => completed(topicList.asJava)
+      case ("createNonPartitionedTopicAsync", _) => onCreate()
+      case ("createPartitionedTopicAsync", _)    => onCreate()
+    }
+    val namespaces: Namespaces = proxy(classOf[Namespaces]) { case ("getNamespacesAsync", _) =>
+      completed(List(namespace.fullyQualified).asJava)
+    }
+    val tenants: Tenants = proxy(classOf[Tenants]) { case ("getTenantsAsync", _) =>
+      completed(List(tenant.tenant).asJava)
+    }
+    proxy(classOf[PulsarAdmin]) {
+      case ("topics", _)     => topics
+      case ("namespaces", _) => namespaces
+      case ("tenants", _)    => tenants
+    }
+  }
+
+  private def clientCreatingWith(onCreate: () => CompletableFuture[Void]): PulsarAdminClient =
+    PulsarAdminClient(mockAdmin(Nil, onCreate))
+
+  override def testSpec: TestSpec =
+    suite("PulsarAdminClientSpec")(
+      suite("topic.createIfDNE - idempotency")(
+        test("succeeds when create returns a 409 PulsarAdminException (concurrent create race)") {
+          val client = clientCreatingWith(() => failed(conflict409))
+          for {
+            exit <- client.topic.createIfDNE(topic, None).exit
+          } yield assert(exit)(succeeds(anything))
+        },
+        test("succeeds when the 409 is wrapped in a CompletionException") {
+          val client = clientCreatingWith(() => failed(new CompletionException(conflict409)))
+          for {
+            exit <- client.topic.createIfDNE(topic, None).exit
+          } yield assert(exit)(succeeds(anything))
+        },
+        test("succeeds via case-insensitive message fallback when status is not 409") {
+          val client = clientCreatingWith(() => failed(new PulsarAdminException("Topic already EXISTS")))
+          for {
+            exit <- client.topic.createIfDNE(topic, None).exit
+          } yield assert(exit)(succeeds(anything))
+        },
+      ),
+      suite("topic.createIfDNE - non-already-exists errors propagate")(
+        test("fails when create returns an unrelated PulsarAdminException") {
+          val client = clientCreatingWith(() => failed(new PulsarAdminException.NotAuthorizedException(new RuntimeException("nope"), "Forbidden", 403)))
+          for {
+            exit <- client.topic.createIfDNE(topic, None).exit
+          } yield assert(exit)(fails(anything))
+        },
+        test("fails when create returns an unrelated RuntimeException") {
+          val client = clientCreatingWith(() => failed(new RuntimeException("kaboom")))
+          for {
+            exit <- client.topic.createIfDNE(topic, None).exit
+          } yield assert(exit)(fails(anything))
+        },
+      ),
+      suite("topic.createIfDNE - preserved behavior")(
+        test("logs a warning and succeeds (without creating) on partition mismatch") {
+          val admin = mockAdmin(
+            topicList = List(topic.ignorePartition.fullyQualified),
+            onCreate = () => failed(new AssertionError("create must not be called when the topic already exists")),
+          )
+          for {
+            exit <- PulsarAdminClient(admin).topic.createIfDNE(topic, Some(3)).exit
+          } yield assert(exit)(succeeds(anything))
+        },
+      ),
+    )
+
+}

report/OXY-146.md

@@ -0,0 +1,50 @@
+# OXY-146 — Pulsar `createIfDNE` idempotency fix
+
+Bug: `createIfDNE` (tenant/namespace/topic) uses check-then-act (list then create) with no
+exception handling. Concurrent callers race; the loser gets a 409 `PulsarAdminException`
+("... already exists"). Fix: treat "already exists" on `create` as success.
+
+File: `modules/events/pulsar/src/main/scala/oxygen/pulsar/client/PulsarAdminClient.scala`
+
+## Decisions / assumptions
+- `PulsarAdminException.getStatusCode` (int) checked first; 409 == already-exists.
+  Verified against pulsar-client-admin-api 4.2.0 jar (status codes; ConflictException = 409).
+- Fallback: case-insensitive `getMessage.contains("already exists")`.
+- Unwrap `CompletionException`/`ExecutionException` (recursively) before predicate, since
+  `ZIO.fromCompletableFuture` can surface the CF's wrapping exception. Guard against null / self cause.
+- Shared private combinator `succeedIfAlreadyExists(logMsg)(create)` = `create.catchSome`.
+  On match: `ZIO.logInfo(logMsg)` and unit. Non-matching errors propagate unchanged.
+- Wrapped the `create(...)` in all three `createIfDNE` methods.
+- Preserved existing behavior: partition-mismatch still logs a warning + succeeds;
+  "already exists via list" fast-path unchanged.
+
+## Test approach
+- No test dir existed for `modules/events/pulsar`. Added `oxygen-test`.jvm % Test dep in build.sbt.
+- Full `PulsarAdmin` (RawPulsarAdminClient) is a large Java interface -> mocked via
+  `java.lang.reflect.Proxy` dispatching by method name (Topics/Namespaces/Tenants sub-clients).
+- Tests via public `topic.createIfDNE`:
+  - 409 ConflictException on create -> succeeds (idempotent).
+  - CompletionException-wrapped 409 -> succeeds (unwrap path).
+  - PulsarAdminException status 500 w/ "Topic already EXISTS" msg -> succeeds (msg fallback, case-insensitive).
+  - unrelated error (403 / RuntimeException) -> fails (propagates).
+
+## Build / test notes
+- `sbt "oxygen-events-pulsar/test"` -> 6 tests pass (idempotency x3, propagation x2, partition-mismatch x1).
+- sbt-git's jgit cannot read a `git worktree` checkout (`NoWorkTreeException`), which breaks project
+  load *in the worktree only*. Worked around during dev with a temporary, uncommitted
+  `zz-worktree-git-fix.sbt` overriding the `git.*` keys; that file was removed and is NOT part of the
+  change set. Building from the main (non-worktree) checkout is unaffected.
+
+## Final summary
+- Implemented exactly as triaged: shared `succeedIfAlreadyExists` combinator wraps `create` in all
+  three `createIfDNE` methods; private `isAlreadyExists` checks 409 -> unwrap Completion/Execution ->
+  case-insensitive "already exists" message.
+- Non-already-exists errors still propagate; partition-mismatch warning + list fast-paths untouched.
+- Added first test dir for the pulsar module (+ `oxygen-test` % Test dep). Mock via JDK dynamic proxy.
+
+CONFIDENCE: 8.5/10
+- High confidence in the production fix and that tests compile + pass against real pulsar 4.2.0 types.
+- Minor uncertainty: exact wrapping behavior of `ZIO.fromCompletableFuture` (whether it unwraps
+  CompletionException itself) is handled defensively either way; the "concurrent race" is modeled by a
+  deterministic single create returning 409 rather than two live fibers (documented, deliberate for
+  test stability).