From 36cd2c42e63199d48520c60ade88ace35f4adefc Mon Sep 17 00:00:00 2001 From: "A.Anikeev" Date: Thu, 13 Aug 2026 11:34:42 +0300 Subject: [PATCH] feat(env-mq-kafka-embedded): create topics on a running environment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Declared topics were already created before anything connects — the convenience constructor passes them to the broker. What was missing is the same guarantee for a topic that appears later, and relying on the broker to create it on demand does not give it: the broker does create the topic, but not in the response to the first metadata request, so a client that assigns partitions right after connecting is left with an empty assignment and every later poll or seek throws. addTopics waits for the creation to complete and fails loudly with the broker's own error, so a spec can add a topic mid-flight and use it immediately. topics() exposes what the environment has, which is what a spec asserts against. Creating topics belongs to the environment, not to whoever consumes it — that is why this lands here rather than in a wait loop inside the Exam tester. Co-Authored-By: Claude Opus 5 (1M context) --- .DS_Store | Bin 0 -> 12292 bytes CLAUDE.md | 64 ++++++++++++++++++ .../mq/kafka/embedded/EmbeddedKafkaSystem.kt | 12 ++++ .../kafka/embedded/EmbeddedKafkaSystemTest.kt | 17 +++++ 4 files changed, 93 insertions(+) create mode 100644 .DS_Store create mode 100644 CLAUDE.md diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..eea98988499b4d62f824866808e4673ed3bbdfaf GIT binary patch literal 12292 zcmeI2TWB0*6vxkKlHFXJtavZDNTnBONGld4Qnn#J_@b2{Snwgc*_dv#+ui1(7L5dH zy&y<^_Q9%%lq$w2jfgD>7OBSIi(;i!5bJ|NK>MIl@qfO#oinpDnf1Ykb|%bxo095oNJ++TVl2b42-Wq+;Xiyoa`HamKFZHnhqRz2+!Af2yZt0#yr5B z%R6boNf+(Zn8L=GBDGB|XNqCj<~W~cInsiYF50PWPL^#>mUCyznW0$b?y$=nIavie zO^;~=(g=hjP{66)z}&k|GgPDrs?ub2O3c19@7@-#uGL3JOVxS_s_gL6*q878`PrWJ zEXplfG+7VGMe}1NVh&ep)KH#ePwfO(B0K3*W@+V&<5NOg-@8_Am7hJwflC7cKO0oX z8p#|@p}Z#T)e_}p=98O}SxFz?-1wARZVB#JtNR|dW~uz^F0g-oJ!Fp>sDZqWR7X0k z{GvB@#jB-1$9_J!5~_!x{G}hsEEGSjG5>X5`5eIWGXwe}ylCJ#s=$XLN~!{H4k}-| za!)6>lzuJW{P|hOuY~uNKDKVH{OvQnY@g?E)s#{Zo^pNUwnLX#=PoX>SiSTu_tS;1 z9N*Tl@oQgF8|6o@@xxzxtYO7*J4&g9dWYYtq_%$80Usi@l7Hlu!BvMpM(?vH(JStu zRiK1erV;ta2uH=8{qDs$t>oUlZy}1?%Ujl>Bznc&GtS)Cd-cYSx0g3!&uw)H$1?tK zRGw_U_m0-ArEhN^-RQP!E!a<{tqs(|9(4X`_H{?34axuQ~qWaevMa4|9*Vv zgj;qo`AdJ2St$OlADF+>D?jdc%HW<=+}RWRnyV!5x!2CSyiM|xobgJ<*B#)SMP#`? z1)hla2Nj>El6=o^yA7{oJ6m-k0{H_baL1k2(&76_VkuFMi+jR@pEF@{v78v`X>Vi;Z?ZJQ@qP%()G` z_7PA>o}5=v zE7bkTj8oZ?}V=u?sA9K6kZ1K}IMRo1Nv(s@}$v-?j ExternalSystem`. `up()` starts all systems concurrently (cached thread pool, one thread per system) with a timeout, rolling back via `down()` on any failure. Look up a started system with `env()` (by type) or `env("NAME")` (by key). + +Two implementation strategies for `ExternalSystem`, both in their own modules: +- **`GenericExternalSystem`** (`env-core`) — wraps any object with `start`/`stop`/`running`/`afterStart` lambdas. Base for non-container systems. +- **`ContainerExternalSystem`** (`env-container`) — wraps a Testcontainers `GenericContainer`. Most `env-db-*` / `env-mq-*` modules build on this. Note: some systems (e.g. `PostgreSqlContainerSystem`) instead *extend* the Testcontainers class directly and implement `ExternalSystem`. + +### The `fixedEnv` flag (important) + +`start(fixedEnv: Boolean)` is the central concept for **fixed vs. dynamic** ports: +- `fixedEnv = true` → bind well-known fixed ports (e.g. Postgres on 5432). Used for stable local dev where the SUT has hardcoded connection config. +- `fixedEnv = false` → let the system pick random free ports. Used in CI / parallel test runs to avoid collisions. + +The value comes from `Environment.Config.envStrategy` — default `SystemPropertyToggle` reads the `ENV_FIXED` system property (default `false`). See `env/example/.../EnvTest.kt` for both modes asserted. + +### Config via system properties + +`Environment.Config` is resolved from system properties by default (`ConfigResolver.FromSystemProperty`): `ENV_FIXED`, `ENV_DRY_RUN`, `ENV_UP_TIMEOUT_SEC` (300), `ENV_DOWN_TIMEOUT_SEC` (10). + +### Standalone runner + +`EnvStarter` (`env-core`) runs any `Environment` subclass as a standalone process (interactive REPL: `s`tatus / `u`p / `d`own / `e`xit), for local dev outside a test run. Invoked via `JavaExec` with the env class name as arg (see README). + +## Module layout + +- `env-core` — interfaces, `Environment`, `EnvStarter`, port utils. No Testcontainers dependency. +- `env-container` — Testcontainers base (`ContainerExternalSystem`, `String.parseImage()` helpers). +- `env-db-*`, `env-mq-*`, `env-redis`, `env-wiremock`, `env-grpc-mock`, `env-selenium`, `env-localstack`, `env-samba`, `env-jar-application` — one system implementation each. Add a new system by creating a `env-` module (register it in `settings.gradle`) and depending on `env-container` or `env-core`. +- `example` — runnable usage reference (`EnvTest.kt`); not published. + +## Conventions + +- **Adding a system**: also add it to `build.gradle`'s `publishMaven` task (the `dependsOn` list) and to `settings.gradle`. +- Config property keys are namespaced `env...` (e.g. `env.mq.rabbit.port`). Define them as constants on the `Config` companion. +- Public API targets Java interop (`@JvmOverloads`, `@JvmStatic`, `@JvmField`); keep that when touching constructors/companions. +- Version lives in one place: `ext.libVersion` in root `build.gradle`. Release commits are just the bumped version number (see git log); publishing is automated on GitHub Release via the `gradle-publish.yml` workflow. +- Style: ktlint + detekt (config in `config/detekt/detekt.yml`), 4-space indent, no trailing commas (`.editorconfig`). diff --git a/env-mq-kafka-embedded/src/main/java/io/github/adven27/env/mq/kafka/embedded/EmbeddedKafkaSystem.kt b/env-mq-kafka-embedded/src/main/java/io/github/adven27/env/mq/kafka/embedded/EmbeddedKafkaSystem.kt index 0d3f54e..9c959cc 100644 --- a/env-mq-kafka-embedded/src/main/java/io/github/adven27/env/mq/kafka/embedded/EmbeddedKafkaSystem.kt +++ b/env-mq-kafka-embedded/src/main/java/io/github/adven27/env/mq/kafka/embedded/EmbeddedKafkaSystem.kt @@ -61,6 +61,18 @@ open class EmbeddedKafkaSystem @JvmOverloads constructor( @Suppress("SpreadOperator") constructor(vararg topics: String) : this(topics = arrayOf(*topics)) + fun addTopics(vararg topics: String) { + embeddedKafka.addTopicsWithResults(*topics) + .filterValues { it != null } + .let { failures -> + require(failures.isEmpty()) { + "Не удалось создать топики: " + failures.entries.joinToString { "${it.key}: ${it.value.message}" } + } + } + } + + fun topics(): Set = embeddedKafka.topics + override fun toString() = "Embedded Kafka Broker" open class Config(val bootstrapServers: String = "PLAINTEXT://localhost:$DEFAULT_KAFKA_PORT") : diff --git a/env-mq-kafka-embedded/src/test/kotlin/io/github/adven27/env/mq/kafka/embedded/EmbeddedKafkaSystemTest.kt b/env-mq-kafka-embedded/src/test/kotlin/io/github/adven27/env/mq/kafka/embedded/EmbeddedKafkaSystemTest.kt index 99b2941..ae3b68b 100644 --- a/env-mq-kafka-embedded/src/test/kotlin/io/github/adven27/env/mq/kafka/embedded/EmbeddedKafkaSystemTest.kt +++ b/env-mq-kafka-embedded/src/test/kotlin/io/github/adven27/env/mq/kafka/embedded/EmbeddedKafkaSystemTest.kt @@ -40,6 +40,23 @@ class EmbeddedKafkaSystemTest { ).use { assertTrue(it.listTopics().names().get().contains("some-topic")) } } + @Test + fun topicsCanBeAddedToARunningEnvironment() { + val env = SomeEnvironment().apply { up() } + sut = env + + env.kafka().addTopics("late-topic") + + assertTrue(env.kafka().topics().contains("late-topic")) + Admin.create( + mapOf( + BOOTSTRAP_SERVERS_CONFIG to env.kafka().config.bootstrapServers, + REQUEST_TIMEOUT_MS_CONFIG to "8000", + DEFAULT_API_TIMEOUT_MS_CONFIG to "12000" + ) + ).use { assertTrue(it.listTopics().names().get().contains("late-topic")) } + } + @Test fun fixedEnvironmentIsNotSupported() { val port = Environment.findAvailableTcpPort()