From f69ee214dba608ad7cb090f968985aa4df6e073e Mon Sep 17 00:00:00 2001 From: Said Boudjelda Date: Thu, 5 Nov 2026 18:03:59 +0100 Subject: [PATCH 1/2] feat(api): idempotent test creation and an sla on flat requests A controller that crashes between POST /api/tests and recording the run id would start a second run on restart, and only a scenario could carry an sla. The Kubernetes operator needs both. POST /api/tests takes an Idempotency-Key, stored in a unique column from the run's first INSERT, so a repeat answers 200 with the run and starts nothing. GET /api/tests?clientKey= finds the run, and /api/health lists the capability. A save that fails leaves the key free, so the retry can start the run once the fault is gone. A flat request takes a top-level sla. One beside a scenario is refused, since the scenario path would otherwise drop it and grade the run without the gate. --- cli/cmd/mcp_caveats.go | 22 +- cli/cmd/mcp_guard.go | 2 +- cli/cmd/mcp_prompts_cluster.go | 2 +- cli/cmd/mcp_tools_runs_assess.go | 2 +- cli/cmd/mcp_tools_runs_list.go | 26 +- cli/cmd/mcp_tools_runs_test.go | 4 +- cli/cmd/mcp_tools_scenario.go | 6 +- cli/schemas/schema_test.go | 5 + cli/schemas/test-request.v1.schema.json | 5 + .../Schemas/test-request.v1.schema.json | 5 + .../schemas/test-request.v1.schema.json | 5 + .../com/bmscomp/kates/api/HealthResource.java | 5 +- .../com/bmscomp/kates/api/TestResource.java | 77 +++++- .../kates/domain/CreateTestRequest.java | 9 + .../com/bmscomp/kates/domain/TestRun.java | 82 ++++-- .../kates/engine/TestOrchestrator.java | 29 ++- .../kates/persistence/EntityMapper.java | 7 +- .../kates/persistence/TestRunEntity.java | 17 ++ .../kates/service/TestRunRepository.java | 11 + .../db/migration/V29__test_run_client_key.sql | 5 + .../bmscomp/kates/api/HealthResourceTest.java | 1 + .../kates/api/TestResourceCreateTest.java | 28 +- .../bmscomp/kates/api/TestResourceTest.java | 245 +++++++++++++++++- .../bmscomp/kates/domain/WireSchemaTest.java | 64 +++++ .../bmscomp/kates/it/IdempotentCreateIT.java | 140 ++++++++++ schemas/test-request.v1.schema.json | 5 + 26 files changed, 736 insertions(+), 73 deletions(-) create mode 100644 kates/src/main/resources/db/migration/V29__test_run_client_key.sql create mode 100644 kates/src/test/java/com/bmscomp/kates/domain/WireSchemaTest.java create mode 100644 kates/src/test/java/com/bmscomp/kates/it/IdempotentCreateIT.java diff --git a/cli/cmd/mcp_caveats.go b/cli/cmd/mcp_caveats.go index bd008573d..e51f5dd62 100644 --- a/cli/cmd/mcp_caveats.go +++ b/cli/cmd/mcp_caveats.go @@ -76,7 +76,7 @@ var mcpCaveatsCore = []mcpCaveat{ Short: "A LOAD run is one producer and one consumer whatever numProducers says; only STRESS runs parallel producers.", Text: "A LOAD run is one producer and one consumer, whatever numProducers and numConsumers say, " + "so it cannot show how the cluster behaves under parallel clients. STRESS starts one producer per numProducers.", - Refs: []string{mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1656-1665", "case LOAD ->", "case STRESS ->")}, + Refs: []string{mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1677-1686", "case LOAD ->", "case STRESS ->")}, }, { ID: mcpCaveatReaperDeadline, @@ -99,11 +99,11 @@ var mcpCaveatsCore = []mcpCaveat{ "kates/src/main/resources/application.properties:333-347", mcpAnchoredRef(mcpJava+"engine/TestTimeoutReaper.java:49-51,67-76,161-164,192-205", "TestResult.TaskStatus.STOPPING", "kates.engine.reaper-grace-ms", "\"-submission\"", "\"Timeout: still \""), - mcpJava + "engine/TestOrchestrator.java:132-154,186-189,354,1125-1184,1250-1259,1314-1346", - mcpJava + "persistence/EntityMapper.java:158-167", + mcpJava + "engine/TestOrchestrator.java:132-163,197-200,366,1142-1205,1271-1280,1335-1367", + mcpJava + "persistence/EntityMapper.java:161-170", "kates/src/main/resources/db/migration/V24__run_planned_duration.sql:1-13", - mcpJava + "domain/TestRun.java:478-481", - mcpJava + "api/HealthResource.java:51-72", + mcpJava + "domain/TestRun.java:520-523", + mcpJava + "api/HealthResource.java:53-75", }, }, { @@ -123,13 +123,13 @@ var mcpCaveatsCore = []mcpCaveat{ "integrity-cg-integrity, without transactions.", Refs: []string{ mcpJava + "domain/TestSpec.java:14-32,111-113", - mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:165,180-182,225", + mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:174,189-191,236", "applyTypeDefaults(type, request.getSpec())", ".withRequestedSpec(", "buildTasks(type, spec, run.getId())"), - mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1045-1091", "TestSpec applyTypeDefaults(", "return merged;"), - mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1635-1639,1730-1734", + mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1062-1108", "TestSpec applyTypeDefaults(", "return merged;"), + mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1656-1660,1751-1755", "Only when asked", `"-integrity-0-group")`), mcpAnchoredRef(mcpJava+"engine/NativeKafkaBackend.java:792", `task.getConsumerGroup() + "-integrity"`), - mcpAnchoredRef(mcpJava+"persistence/EntityMapper.java:56,84-85,143-145", + mcpAnchoredRef(mcpJava+"persistence/EntityMapper.java:56,85-86,146-148", "entity.setRequestedSpecJson(toJson(run.getRequestedSpec()))", "fromJson(entity.getRequestedSpecJson()", "if (run.getRequestedSpec() != null)"), "kates/src/main/resources/db/migration/V23__requested_spec_fields.sql:1-6", @@ -174,7 +174,7 @@ var mcpCaveatsCore = []mcpCaveat{ Text: "A TUNE_* run executes one produce task with the spec's single configuration: it measures one setting " + "and compares none.", Refs: []string{ - mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1741-1742", "case TUNE_REPLICATION,", "-tune-0"), + mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1762-1763", "case TUNE_REPLICATION,", "-tune-0"), mcpJava + "trogdor/SpecFactory.java:38-39", }, }, @@ -186,7 +186,7 @@ var mcpCaveatsCore = []mcpCaveat{ Refs: []string{ mcpJava + "trend/TrendResource.java:31-60", mcpJava + "trend/TrendService.java:53-54,185-189", - mcpAnchoredRef(mcpJava+"service/TestRunRepository.java:341-361", + mcpAnchoredRef(mcpJava+"service/TestRunRepository.java:351-371", "List findWithResults(", "WHERE r.testType = :type AND r.status = :status", ".map(EntityMapper::toDomain)"), }, diff --git a/cli/cmd/mcp_guard.go b/cli/cmd/mcp_guard.go index 871bd09d0..9d45ca15f 100644 --- a/cli/cmd/mcp_guard.go +++ b/cli/cmd/mcp_guard.go @@ -125,7 +125,7 @@ func mcpInvalidArgument(message, value string) error { // mcpID is a test-run or disruption id as tool input. The backend mints both // as the first 8 characters of a random UUID, so 8 lowercase hex characters -// (domain/TestRun.java:46,64; disruption/DisruptionLauncher.java:77). The +// (domain/TestRun.java:48,67; disruption/DisruptionLauncher.java:77). The // schema carries the pattern, so the SDK refuses a bad id before any Kates // code runs; mcpValidateID checks again for ids that arrive another way. type mcpID string diff --git a/cli/cmd/mcp_prompts_cluster.go b/cli/cmd/mcp_prompts_cluster.go index 13365b1dd..faefd0390 100644 --- a/cli/cmd/mcp_prompts_cluster.go +++ b/cli/cmd/mcp_prompts_cluster.go @@ -95,7 +95,7 @@ func mcpDidKatesCauseThis(_ context.Context, req *mcp.GetPromptRequest) (*mcp.Ge // mcpDidKatesCauseThisText is the prompt itself. Every fact it states about // Kates was checked against the code: runs and disruptions record no owner -// (domain/TestRun.java:16-42, disruption/DisruptionReportEntity.java:15-34); +// (domain/TestRun.java:16-44, disruption/DisruptionReportEntity.java:15-34); // audit rows name the principal that made each change (V26), and every REST // and gRPC change and every scheduled firing writes one // (audit/AuditResponseFilter.java, security/GrpcApiKeyInterceptor.java, diff --git a/cli/cmd/mcp_tools_runs_assess.go b/cli/cmd/mcp_tools_runs_assess.go index c5e8cd3b4..f42b8b3dd 100644 --- a/cli/cmd/mcp_tools_runs_assess.go +++ b/cli/cmd/mcp_tools_runs_assess.go @@ -361,7 +361,7 @@ func mcpRunHasRequested(r *client.MCPRun) bool { // mcpAssessBaseline finds the baseline run: among the runs the band's scan // read, or else with one read of its own. The baseline may be any run, of -// any type or status (TestResource.java:590-618 checks only that it exists), +// any type or status (TestResource.java:659-687 checks only that it exists), // so reading it polls it if it is still active, as get_run does. func mcpAssessBaseline(ctx context.Context, call *mcpCall, baselineID string, band mcpBandScan) *mcpRunIdentity { if b, ok := band.byID[baselineID]; ok { diff --git a/cli/cmd/mcp_tools_runs_list.go b/cli/cmd/mcp_tools_runs_list.go index f5cea2480..c3b2b11b8 100644 --- a/cli/cmd/mcp_tools_runs_list.go +++ b/cli/cmd/mcp_tools_runs_list.go @@ -26,7 +26,7 @@ import ( // Every tool goes through addReadTool and reads only with GET. get_run and // assess_run read the run with GET /api/tests/{id}, which makes the backend // poll a run that is still active and save what it finds -// (TestResource.java:458-464, TestOrchestrator.refreshStatus); both +// (TestResource.java:527-533, TestOrchestrator.refreshStatus); both // descriptions say so. Third-party text (task errors, scenario and phase // names, labels, plan names, audit details, backend messages) is fenced; // ids, types and statuses are cleaned so an agent can pass them back. @@ -574,15 +574,15 @@ func mcpRunDecodeSpec(raw json.RawMessage, w *mcpRunSpecWire) (bool, error) { } // mcpRunSpecFrom reads the stored spec. A run without a topic used one named -// after its type (TestOrchestrator.java:1614,1941-1942). The topic, acks and +// after its type (TestOrchestrator.java:1635,1962-1963). The topic, acks and // compressionType are shown as identifiers only when they hold values Kafka // accepts (mcpKafkaTopicRE, mcpAcksRE, mcpCompressionRE, the values // TestSpec's bean validation accepts). A stored run can hold others: a // schedule saved by a PUT, which runs no bean validation // (ScheduleResource.java:224-226), fires its spec as sent, a Kates API -// without the check of a scenario's specs (TestOrchestrator.java:1126-1130) -// stored a scenario's base spec as sent (TestOrchestrator.java:347-355, -// 373-377), and one without the check of a gRPC request's fields +// without the check of a scenario's specs (TestOrchestrator.java:1143-1147) +// stored a scenario's base spec as sent (TestOrchestrator.java:358-367, +// 385-389), and one without the check of a gRPC request's fields // (GrpcTestService.java:116-121) stored a compressionType sent over gRPC // unchecked. So another value is third-party text and goes, fenced, into // invalid. The seven fields a backend without @@ -706,7 +706,7 @@ func mcpRunRequestedFrom(call *mcpCall, run *client.MCPRun) (*mcpRunRequestedSpe func mcpRunTaskFrom(call *mcpCall, t client.MCPRunTask) mcpRunTaskOut { return mcpRunTaskOut{ // A scenario run's task ids are the run id and the phase name - // (TestOrchestrator.java:1758), which nothing validates. + // (TestOrchestrator.java:1779), which nothing validates. TaskID: call.FenceN(t.TaskID, 128), Phase: call.FenceN(t.PhaseName, 64), Status: mcpSanitizeLine(t.Status, 16), @@ -1031,7 +1031,7 @@ var mcpCaveatsRunsList = []mcpCaveat{ mcpAnchoredRef(mcpJava+"report/ReportGenerator.java:220-249,283-294", "report.setSummary(MetricUtils.computeSummary(results));", "summaries.put(phase, MetricUtils.computeSummary(rows))"), - mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:415-424,2089-2105", + mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:427-436,2110-2126", "TestResult.TaskStatus.PENDING : TestResult.TaskStatus.RUNNING", "private static TestRun withUnfinishedTasksFailed("), }, @@ -1051,7 +1051,7 @@ var mcpCaveatsRunsList = []mcpCaveat{ "private void runIntegrity(", "state.integrityResult = state.verifier.verify("), mcpAnchoredRef(mcpJava+"engine/DataIntegrityVerifier.java:51,224-227", "static final int MAX_LOST_RANGES = 1000;", "if (chaosStartNanos <= 0) {"), - mcpAnchoredRef(mcpJava+"persistence/EntityMapper.java:34-43,216-223,243", + mcpAnchoredRef(mcpJava+"persistence/EntityMapper.java:34-43,219-226,246", "entity.setIntegrityJson(integrityJson);", ".withIntegrity(fromJson(INTEGRITY_JSON, entity.getIntegrityJson(), IntegrityResult.class));", "DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false"), @@ -1070,9 +1070,9 @@ var mcpCaveatsRunsList = []mcpCaveat{ "except for a run cancelled before its tasks existed; a cancel through the REST API also leaves a CANCEL " + "audit row.", Refs: []string{ - mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:2040-2118", + mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:2061-2139", `"Cancelled by user"`, `EventKind.FAILED, "cancelled"`, "return run.withResults(updatedResults);"), - mcpAnchoredRef(mcpJava+"api/TestResource.java:497-542", + mcpAnchoredRef(mcpJava+"api/TestResource.java:566-611", `@Path("/{id}/cancel")`, `auditService.record("CANCEL"`, `"Test cancelled; it is stored as FAILED"`), mcpJava + "domain/TestResult.java:25-31", }, @@ -1100,14 +1100,14 @@ var mcpCaveatsRunsList = []mcpCaveat{ "say which the API is, but a scenario run's tasks do: on an older API they all start within moments " + "of one another.", Refs: []string{ - mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:341-343,391-402,404-420", + mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:352-354,403-414,416-432", "applyTypeDefaults(type, scenario.getBaseSpec())", ".withRequestedSpec(", "long phaseStartMs = System.currentTimeMillis();", "scenario.resolveSpecForPhase(phase)", "phaseStartMs = saturatedSum(phaseStartMs", "backend.submit(task)", "TestResult.TaskStatus.PENDING"), - mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1452-1480,1552-1591,1601-1613", + mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1473-1501,1573-1612,1622-1634", "MAX_RAMP_STEPS = 100", "scenarioInapplicableFields(TestScenario scenario", "is a SPIKE phase", "has none (", "check no record CRCs"), - mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1738-1807", "List buildPhaseTask(", + mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1759-1828", "List buildPhaseTask(", "int baseTarget = Math.max(1, spec.getThroughput() / steps);", "saturatedSum(startAtMs, s * stepMs)", `taskId + "-spike"`), mcpAnchoredRef(mcpJava+"domain/TestScenario.java:108-183", diff --git a/cli/cmd/mcp_tools_runs_test.go b/cli/cmd/mcp_tools_runs_test.go index b73dce741..814780181 100644 --- a/cli/cmd/mcp_tools_runs_test.go +++ b/cli/cmd/mcp_tools_runs_test.go @@ -813,9 +813,9 @@ func mcpOutsideFences(h *mcpHarness, raw []byte) string { } // A scenario run's text reaches the store unvalidated: its phase names -// become task ids (TestOrchestrator.java:1758), and an older Kates API +// become task ids (TestOrchestrator.java:1779), and an older Kates API // stored its base spec's topic, acks and compression as sent -// (TestOrchestrator.java:347-355,373-377). None of it may reach the model outside a +// (TestOrchestrator.java:358-367,385-389). None of it may reach the model outside a // fence. func TestMCPGetRunUnvalidatedScenarioText(t *testing.T) { fb := newMCPFakeBackend(t, "cluster-a") diff --git a/cli/cmd/mcp_tools_scenario.go b/cli/cmd/mcp_tools_scenario.go index 14d78d4c6..a71bbf247 100644 --- a/cli/cmd/mcp_tools_scenario.go +++ b/cli/cmd/mcp_tools_scenario.go @@ -1240,9 +1240,9 @@ var mcpCaveatsScenario = []mcpCaveat{ "and kates test apply sends a scenario with whatever key it holds.", Refs: []string{ "plans/mcp-server.md:387-405", - mcpJava + "api/TestResource.java:108-135", + mcpJava + "api/TestResource.java:124-192", mcpJava + "domain/TestSpec.java:34-113", - mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1372-1447", + mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1393-1468", "Map inapplicableFields(", "the trogdor backend cannot run a transactional producer"), mcpAnchoredRef("cli/cmd/apply.go:238-307", "for i, scenario := range sf.Scenarios {", "apiClient.CreateTest(a.ctx, scenarioToRequest(scenario))", @@ -1259,7 +1259,7 @@ var mcpCaveatsScenario = []mcpCaveat{ "listed as defaulted may differ on the cluster that runs the scenario.", Refs: []string{ mcpJava + "config/TestTypeDefaults.java:22-64,323-464", - mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1045-1091", "TestSpec applyTypeDefaults(", "return merged;"), + mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1062-1108", "TestSpec applyTypeDefaults(", "return merged;"), mcpAnchoredRef("kates/src/main/resources/application.properties:73-114", "# Per-type overrides: STRESS", "kates.tests.roundtrip.throughput=10000"), mcpAnchoredRef("charts/kates/values.yaml:691-759", "defaults:", "tests:", `throughput: "10000"`), diff --git a/cli/schemas/schema_test.go b/cli/schemas/schema_test.go index 6a479643b..115d5f638 100644 --- a/cli/schemas/schema_test.go +++ b/cli/schemas/schema_test.go @@ -411,6 +411,11 @@ func TestSchemasRejectInvalidEnumsTypesAndBounds(t *testing.T) { schema: "test-request.v1.schema.json", doc: `{"scenario": {"name": "s", "phases": [{"name": "p1", "type": "RAMP", "durationMs": 1000, "rampSteps": 1}]}}`, }, + { + name: "test-request sla beside a scenario", + schema: "test-request.v1.schema.json", + doc: `{"sla": {"maxP99LatencyMs": 1}, "scenario": {"name": "s", "phases": [{"name": "p1", "type": "STEADY", "durationMs": 1000}]}}`, + }, { name: "flow invalid action enum", schema: "flow.v1.schema.json", diff --git a/cli/schemas/test-request.v1.schema.json b/cli/schemas/test-request.v1.schema.json index 759d347fd..e5f9741a4 100644 --- a/cli/schemas/test-request.v1.schema.json +++ b/cli/schemas/test-request.v1.schema.json @@ -9,6 +9,7 @@ { "required": ["type"] }, { "required": ["scenario"] } ], + "not": { "required": ["sla", "scenario"] }, "properties": { "type": { "description": "Test type for a single-phase run (or fallback if scenario.type is omitted).", @@ -22,6 +23,10 @@ "description": "Wire-format TestSpec parameters (uses numRecords, numProducers, recordSize, durationMs, throughput).", "$ref": "#/$defs/wireTestSpec" }, + "sla": { + "description": "Pass/fail criteria for a run with no scenario; a scenario carries its own under scenario.sla.", + "$ref": "#/$defs/slaDefinition" + }, "scenario": { "description": "Multi-phase scenario envelope with baseSpec, ordered execution phases, SLA definition, and metadata labels.", "$ref": "#/$defs/multiPhaseScenario" diff --git a/editors/visualstudio-kates/Schemas/test-request.v1.schema.json b/editors/visualstudio-kates/Schemas/test-request.v1.schema.json index 759d347fd..e5f9741a4 100644 --- a/editors/visualstudio-kates/Schemas/test-request.v1.schema.json +++ b/editors/visualstudio-kates/Schemas/test-request.v1.schema.json @@ -9,6 +9,7 @@ { "required": ["type"] }, { "required": ["scenario"] } ], + "not": { "required": ["sla", "scenario"] }, "properties": { "type": { "description": "Test type for a single-phase run (or fallback if scenario.type is omitted).", @@ -22,6 +23,10 @@ "description": "Wire-format TestSpec parameters (uses numRecords, numProducers, recordSize, durationMs, throughput).", "$ref": "#/$defs/wireTestSpec" }, + "sla": { + "description": "Pass/fail criteria for a run with no scenario; a scenario carries its own under scenario.sla.", + "$ref": "#/$defs/slaDefinition" + }, "scenario": { "description": "Multi-phase scenario envelope with baseSpec, ordered execution phases, SLA definition, and metadata labels.", "$ref": "#/$defs/multiPhaseScenario" diff --git a/editors/vscode-kates/schemas/test-request.v1.schema.json b/editors/vscode-kates/schemas/test-request.v1.schema.json index 759d347fd..e5f9741a4 100644 --- a/editors/vscode-kates/schemas/test-request.v1.schema.json +++ b/editors/vscode-kates/schemas/test-request.v1.schema.json @@ -9,6 +9,7 @@ { "required": ["type"] }, { "required": ["scenario"] } ], + "not": { "required": ["sla", "scenario"] }, "properties": { "type": { "description": "Test type for a single-phase run (or fallback if scenario.type is omitted).", @@ -22,6 +23,10 @@ "description": "Wire-format TestSpec parameters (uses numRecords, numProducers, recordSize, durationMs, throughput).", "$ref": "#/$defs/wireTestSpec" }, + "sla": { + "description": "Pass/fail criteria for a run with no scenario; a scenario carries its own under scenario.sla.", + "$ref": "#/$defs/slaDefinition" + }, "scenario": { "description": "Multi-phase scenario envelope with baseSpec, ordered execution phases, SLA definition, and metadata labels.", "$ref": "#/$defs/multiPhaseScenario" diff --git a/kates/src/main/java/com/bmscomp/kates/api/HealthResource.java b/kates/src/main/java/com/bmscomp/kates/api/HealthResource.java index fd004af1e..a691f34fb 100644 --- a/kates/src/main/java/com/bmscomp/kates/api/HealthResource.java +++ b/kates/src/main/java/com/bmscomp/kates/api/HealthResource.java @@ -1,6 +1,7 @@ package com.bmscomp.kates.api; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; import jakarta.annotation.security.PermitAll; import jakarta.inject.Inject; @@ -47,12 +48,14 @@ public HealthResource( @GET @Operation( summary = "Application health", - description = "Returns engine status, Kafka connectivity, and test type configurations") + description = "Returns engine status, Kafka connectivity, test type configurations, and the" + + " capabilities a client can check before relying on them") public Map health() { boolean kafkaReachable = clusterHealthService.isReachable(); Map response = new LinkedHashMap<>(); response.put("status", kafkaReachable ? "UP" : "DEGRADED"); + response.put("capabilities", List.of("clientKey")); response.put( "engine", diff --git a/kates/src/main/java/com/bmscomp/kates/api/TestResource.java b/kates/src/main/java/com/bmscomp/kates/api/TestResource.java index ce960a347..038ad7cf6 100644 --- a/kates/src/main/java/com/bmscomp/kates/api/TestResource.java +++ b/kates/src/main/java/com/bmscomp/kates/api/TestResource.java @@ -5,7 +5,9 @@ import java.util.EnumSet; import java.util.List; import java.util.Locale; +import java.util.Map; import java.util.Set; +import java.util.stream.Stream; import jakarta.annotation.security.RolesAllowed; import jakarta.inject.Inject; import jakarta.validation.Valid; @@ -19,11 +21,14 @@ import jakarta.ws.rs.PathParam; import jakarta.ws.rs.Produces; import jakarta.ws.rs.QueryParam; +import jakarta.ws.rs.core.Context; +import jakarta.ws.rs.core.HttpHeaders; import jakarta.ws.rs.core.MediaType; import jakarta.ws.rs.core.Response; import io.smallrye.common.annotation.Blocking; import org.eclipse.microprofile.openapi.annotations.Operation; +import org.eclipse.microprofile.openapi.annotations.enums.ParameterIn; import org.eclipse.microprofile.openapi.annotations.enums.SchemaType; import org.eclipse.microprofile.openapi.annotations.media.Content; import org.eclipse.microprofile.openapi.annotations.media.Schema; @@ -73,6 +78,17 @@ public class TestResource { */ private static final int PRUNE_LIMIT = 1000; + /** What test_runs.client_key holds. */ + private static final int CLIENT_KEY_LENGTH = 128; + + /** + * Serialises the creates of one Idempotency-Key, so that the second finds + * the first's run. Enough because the Kates API runs one replica; beyond + * that the unique index holds. Striped, so no map of keys has to be emptied. + */ + private static final Object[] CLIENT_KEY_LOCKS = + Stream.generate(Object::new).limit(64).toArray(); + private static final String AN_INSTANT = "an ISO-8601 instant such as 2026-09-07T00:00:00Z"; private static final Logger LOG = Logger.getLogger(TestResource.class); @@ -109,17 +125,58 @@ public TestResource( @Operation( summary = "Create and execute a test", description = "Submits a new performance test run for asynchronous execution") + @APIResponse(responseCode = "200", description = "The run this Idempotency-Key already created; nothing started") @APIResponse(responseCode = "202", description = "Test accepted for execution") @APIResponse( responseCode = "400", description = "Invalid request, including a value outside its limits in the spec or a scenario's specs," - + " or a spec field the test type or backend cannot apply; fieldErrors names each field") + + " a spec field the test type or backend cannot apply, an sla beside a scenario, or an" + + " Idempotency-Key that is blank, over 128 characters or holds a control character;" + + " fieldErrors names each field") @APIResponse(responseCode = "429", description = "Concurrency limit reached — retry later") @APIResponse( responseCode = "500", description = "A fault in the Kates API, such as its database being unreachable; the run did not start") - public Response createTest(@Valid CreateTestRequest request) { - var result = orchestrator.executeTest(request); + @Parameter( + name = "Idempotency-Key", + in = ParameterIn.HEADER, + description = "Starts at most one run for this key, and answers later requests with it", + schema = @Schema(type = SchemaType.STRING, maxLength = CLIENT_KEY_LENGTH)) + public Response createTest(@Valid CreateTestRequest request, @Context HttpHeaders headers) { + // Not @HeaderParam: it reads a blank header as absent, which would + // start a run the caller meant to key. + return create(request, headers.getHeaderString("Idempotency-Key")); + } + + private Response create(CreateTestRequest request, String clientKey) { + if (clientKey == null) { + return start(request, null); + } + if (!isClientKey(clientKey)) { + String refused = "Idempotency-Key must be 1 to " + CLIENT_KEY_LENGTH + + " characters, not blank, with no control characters"; + return Response.status(400) + .entity(ApiError.validationFailed(refused, Map.of("Idempotency-Key", refused))) + .build(); + } + synchronized (CLIENT_KEY_LOCKS[Math.floorMod(clientKey.hashCode(), CLIENT_KEY_LOCKS.length)]) { + // A replay starts nothing, so it leaves only the response filter's row. + return repository + .findByClientKey(clientKey) + .map(run -> Response.ok(run).build()) + .orElseGet(() -> start(request, clientKey)); + } + } + + /** Whether a run can hold this key: POST refuses any other, and Postgres refuses NUL in a query. */ + private static boolean isClientKey(String key) { + return !key.isBlank() + && key.length() <= CLIENT_KEY_LENGTH + && key.chars().noneMatch(Character::isISOControl); + } + + private Response start(CreateTestRequest request, String clientKey) { + var result = orchestrator.executeTest(request, clientKey); if (result.isFailure()) { Exception failure = result.asFailure().orElseThrow(); ApiError error = notStarted(request, failure); @@ -399,17 +456,29 @@ private static Response badRequest(String message) { @GET @Operation( summary = "List test runs", - description = "Returns paginated test runs, optionally filtered by type or status") + description = "Returns paginated test runs, optionally filtered by type or status, or the run an" + + " Idempotency-Key created") @APIResponse(responseCode = "200", description = "Paginated list of test runs") public Response listTests( @Parameter(description = "Filter by test type") @QueryParam("type") String type, @Parameter(description = "Filter by status") @QueryParam("status") String status, + @Parameter(description = "The Idempotency-Key the run was created with; type and status are ignored") + @QueryParam("clientKey") + String clientKey, @Parameter(description = "Page number (0-based)") @QueryParam("page") @DefaultValue("0") int page, @Parameter(description = "Page size (max 200)") @QueryParam("size") @DefaultValue("50") int size) { int safePage = Math.max(0, page); int safeSize = Math.max(1, Math.min(size, 200)); + if (clientKey != null && !clientKey.isEmpty()) { + List content = isClientKey(clientKey) + ? repository.findByClientKey(clientKey).stream().toList() + : List.of(); + return Response.ok(new PagedResponse<>(content, safePage, safeSize, content.size())) + .build(); + } + if (type != null && !type.isEmpty()) { try { TestType testType = TestType.valueOf(type.toUpperCase()); diff --git a/kates/src/main/java/com/bmscomp/kates/domain/CreateTestRequest.java b/kates/src/main/java/com/bmscomp/kates/domain/CreateTestRequest.java index 01bc3e8a4..a2c7c61b7 100644 --- a/kates/src/main/java/com/bmscomp/kates/domain/CreateTestRequest.java +++ b/kates/src/main/java/com/bmscomp/kates/domain/CreateTestRequest.java @@ -16,6 +16,7 @@ public class CreateTestRequest { private String backend; private TestScenario scenario; + private SlaDefinition sla; public CreateTestRequest() {} @@ -51,6 +52,14 @@ public void setScenario(TestScenario scenario) { this.scenario = scenario; } + public SlaDefinition getSla() { + return sla; + } + + public void setSla(SlaDefinition sla) { + this.sla = sla; + } + public boolean isScenario() { return scenario != null && scenario.getPhases() != null diff --git a/kates/src/main/java/com/bmscomp/kates/domain/TestRun.java b/kates/src/main/java/com/bmscomp/kates/domain/TestRun.java index 2f36f3a11..a0b010b00 100644 --- a/kates/src/main/java/com/bmscomp/kates/domain/TestRun.java +++ b/kates/src/main/java/com/bmscomp/kates/domain/TestRun.java @@ -40,6 +40,8 @@ public class TestRun { * not part of the run the API answers with. */ private final Long plannedDurationMs; + /** The Idempotency-Key the run was created with; null when it had none. */ + private final String clientKey; public TestRun() { this( @@ -56,6 +58,7 @@ public TestRun() { null, null, null, + null, null); } @@ -74,6 +77,7 @@ public TestRun(TestType testType, TestSpec spec) { null, null, null, + null, null); } @@ -91,7 +95,8 @@ private TestRun( Map cdcPhases, String cdcPhase, Map requestedSpec, - Long plannedDurationMs) { + Long plannedDurationMs, + String clientKey) { this.id = id; this.testType = testType; this.spec = spec; @@ -106,6 +111,7 @@ private TestRun( this.cdcPhase = cdcPhase; this.requestedSpec = requestedSpec != null ? new LinkedHashMap<>(requestedSpec) : null; this.plannedDurationMs = plannedDurationMs; + this.clientKey = clientKey; } public TestRun withResult(TestResult result) { @@ -125,7 +131,8 @@ public TestRun withResult(TestResult result) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withId(String id) { @@ -143,7 +150,8 @@ public TestRun withId(String id) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withTestType(TestType testType) { @@ -161,7 +169,8 @@ public TestRun withTestType(TestType testType) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withSpec(TestSpec spec) { @@ -179,7 +188,8 @@ public TestRun withSpec(TestSpec spec) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withStatus(TestResult.TaskStatus status) { @@ -197,7 +207,8 @@ public TestRun withStatus(TestResult.TaskStatus status) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withResults(List results) { @@ -215,7 +226,8 @@ public TestRun withResults(List results) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withCreatedAt(String createdAt) { @@ -233,7 +245,8 @@ public TestRun withCreatedAt(String createdAt) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withBackend(String backend) { @@ -251,7 +264,8 @@ public TestRun withBackend(String backend) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withScenarioName(String scenarioName) { @@ -269,7 +283,8 @@ public TestRun withScenarioName(String scenarioName) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withLabels(Map labels) { @@ -287,7 +302,8 @@ public TestRun withLabels(Map labels) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withSla(SlaDefinition sla) { @@ -305,7 +321,8 @@ public TestRun withSla(SlaDefinition sla) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withCdcPhases(Map cdcPhases) { @@ -323,7 +340,8 @@ public TestRun withCdcPhases(Map cdcPhases) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withCdcPhase(String cdcPhase) { @@ -341,7 +359,8 @@ public TestRun withCdcPhase(String cdcPhase) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withRequestedSpec(Map requestedSpec) { @@ -359,7 +378,8 @@ public TestRun withRequestedSpec(Map requestedSpec) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withPlannedDurationMs(Long plannedDurationMs) { @@ -377,7 +397,27 @@ public TestRun withPlannedDurationMs(Long plannedDurationMs) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); + } + + public TestRun withClientKey(String clientKey) { + return new TestRun( + id, + testType, + spec, + status, + results, + createdAt, + backend, + scenarioName, + labels, + sla, + cdcPhases, + cdcPhase, + requestedSpec, + plannedDurationMs, + clientKey); } public TestRun withAddedResult(TestResult result) { @@ -398,7 +438,8 @@ public TestRun withAddedResult(TestResult result) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public TestRun withUpdatedResult(TestResult updatedResult) { @@ -420,7 +461,8 @@ public TestRun withUpdatedResult(TestResult updatedResult) { cdcPhases, cdcPhase, requestedSpec, - plannedDurationMs); + plannedDurationMs, + clientKey); } public String getId() { @@ -479,4 +521,8 @@ public Map getRequestedSpec() { public Long getPlannedDurationMs() { return plannedDurationMs; } + + public String getClientKey() { + return clientKey; + } } diff --git a/kates/src/main/java/com/bmscomp/kates/engine/TestOrchestrator.java b/kates/src/main/java/com/bmscomp/kates/engine/TestOrchestrator.java index 84779c4ef..ba8e586fb 100644 --- a/kates/src/main/java/com/bmscomp/kates/engine/TestOrchestrator.java +++ b/kates/src/main/java/com/bmscomp/kates/engine/TestOrchestrator.java @@ -149,8 +149,17 @@ void recoverOrphans() { } public com.bmscomp.kates.util.Result executeTest(CreateTestRequest request) { + return executeTest(request, null); + } + + /** + * Starts the run with {@code clientKey} in its first INSERT, so that + * two creates with one key cannot both store a run; the column's unique + * index fails the second as any failed save, permit released. + */ + public com.bmscomp.kates.util.Result executeTest(CreateTestRequest request, String clientKey) { if (request.isScenario()) { - return executeScenario(request); + return executeScenario(request, clientKey); } // Before the permit: a request that cannot run as written is the @@ -180,7 +189,9 @@ public com.bmscomp.kates.util.Result executeTest(CreateTestR TestRun run = new TestRun(type, spec) .withBackend(backendName) .withRequestedSpec(explicitFieldsOf(request.getSpec())) - .withPlannedDurationMs(plannedDurationMs(type, spec)); + .withSla(request.getSla()) + .withPlannedDurationMs(plannedDurationMs(type, spec)) + .withClientKey(clientKey); // Register BEFORE the first thing that can throw. A transient failure in // save/fireEvent used to strand the permit forever (the semaphore drained // one permit per failure until restart), because nothing had recorded @@ -324,7 +335,7 @@ void executeAsync(TestRun run, TestType type, TestSpec spec, String backendName, * not a staircase. */ @io.opentelemetry.instrumentation.annotations.WithSpan("TestOrchestrator.executeScenario") - com.bmscomp.kates.util.Result executeScenario(CreateTestRequest request) { + com.bmscomp.kates.util.Result executeScenario(CreateTestRequest request, String clientKey) { TestScenario scenario = request.getScenario(); TestType type = scenario.getType() != null ? scenario.getType() : request.getType(); String backendName = scenarioBackend(request); @@ -346,6 +357,7 @@ com.bmscomp.kates.util.Result executeScenario(CreateTestRequ .withLabels(scenario.getLabels()) .withSla(scenario.getSla()) .withPlannedDurationMs(plannedDurationMs(scenario)) + .withClientKey(clientKey) .withStatus(TestResult.TaskStatus.RUNNING); // Scenarios previously bypassed the concurrency cap entirely — executeTest @@ -721,10 +733,15 @@ public TestRun refreshStatus(String runId) { * it were about to start until the timeout reaper failed it, once its * deadline had passed. Failing to write this is not worth * masking the original error, so it only logs. + * + *

Saved without its key: a first save that failed left no row, and this + * one would hold the key for a run that never started, which every retry + * would then be answered with. A row that did commit keeps its key, since + * no save updates that column. */ private void markStrandedAsFailed(TestRun run) { try { - repository.save(run.withStatus(TestResult.TaskStatus.FAILED)); + repository.save(run.withClientKey(null).withStatus(TestResult.TaskStatus.FAILED)); } catch (RuntimeException e) { LOG.warnf("Could not mark stranded run %s as FAILED: %s", run.getId(), e.getMessage()); } @@ -1154,6 +1171,10 @@ public java.util.Optional refusal(CreateTestRequest re new InvalidTestSpecException("", Map.of("type", why + " to one of " + typeNames()))); } if (request.isScenario()) { + if (request.getSla() != null) { + return java.util.Optional.of(new InvalidTestSpecException( + "", Map.of("sla", "a scenario carries its own; move it to scenario.sla"))); + } TestScenario scenario = request.getScenario(); // The checks below read every phase, so a null one threw there. Map nulls = new java.util.LinkedHashMap<>(); diff --git a/kates/src/main/java/com/bmscomp/kates/persistence/EntityMapper.java b/kates/src/main/java/com/bmscomp/kates/persistence/EntityMapper.java index 57c92e9c5..9934a00a1 100644 --- a/kates/src/main/java/com/bmscomp/kates/persistence/EntityMapper.java +++ b/kates/src/main/java/com/bmscomp/kates/persistence/EntityMapper.java @@ -58,6 +58,7 @@ public static TestRunEntity toEntity(TestRun run) { entity.setLabelsJson(toJson(run.getLabels())); entity.setCdcPhasesJson(toJson(run.getCdcPhases())); entity.setPlannedDurationMs(run.getPlannedDurationMs()); + entity.setClientKey(run.getClientKey()); if (run.getResults() != null) { for (TestResult result : run.getResults()) { @@ -83,7 +84,8 @@ public static TestRun toDomain(TestRunEntity entity) { .withCdcPhases(fromJson(entity.getCdcPhasesJson(), new TypeReference>() {})) .withRequestedSpec( fromJson(entity.getRequestedSpecJson(), new TypeReference>() {})) - .withPlannedDurationMs(entity.getPlannedDurationMs()); + .withPlannedDurationMs(entity.getPlannedDurationMs()) + .withClientKey(entity.getClientKey()); if (entity.getResults() != null) { run = run.withResults(entity.getResults().stream() @@ -114,7 +116,8 @@ public static TestRun toDomainSummary(TestRunEntity entity) { .withCdcPhases(fromJson(entity.getCdcPhasesJson(), new TypeReference>() {})) .withRequestedSpec( fromJson(entity.getRequestedSpecJson(), new TypeReference>() {})) - .withPlannedDurationMs(entity.getPlannedDurationMs()); + .withPlannedDurationMs(entity.getPlannedDurationMs()) + .withClientKey(entity.getClientKey()); } /** diff --git a/kates/src/main/java/com/bmscomp/kates/persistence/TestRunEntity.java b/kates/src/main/java/com/bmscomp/kates/persistence/TestRunEntity.java index 9a13f6cf6..6cbcf601e 100644 --- a/kates/src/main/java/com/bmscomp/kates/persistence/TestRunEntity.java +++ b/kates/src/main/java/com/bmscomp/kates/persistence/TestRunEntity.java @@ -82,6 +82,15 @@ public class TestRunEntity { @Column(name = "planned_duration_ms") private Long plannedDurationMs; + /** + * The Idempotency-Key the run was created with. Unique here as in V29, + * since the unit tests build their schema from this entity, and never + * updated: a save mutates the managed entity, from copies that may not + * carry it. + */ + @Column(name = "client_key", length = 128, unique = true, updatable = false) + private String clientKey; + @OneToMany(mappedBy = "testRun", cascade = CascadeType.ALL, orphanRemoval = true, fetch = FetchType.LAZY) @OrderBy("id ASC") private List results = new ArrayList<>(); @@ -189,6 +198,14 @@ public void setPlannedDurationMs(Long plannedDurationMs) { this.plannedDurationMs = plannedDurationMs; } + public String getClientKey() { + return clientKey; + } + + public void setClientKey(String clientKey) { + this.clientKey = clientKey; + } + public List getResults() { return results; } diff --git a/kates/src/main/java/com/bmscomp/kates/service/TestRunRepository.java b/kates/src/main/java/com/bmscomp/kates/service/TestRunRepository.java index 0a350584f..74f044201 100644 --- a/kates/src/main/java/com/bmscomp/kates/service/TestRunRepository.java +++ b/kates/src/main/java/com/bmscomp/kates/service/TestRunRepository.java @@ -105,6 +105,17 @@ public Optional findById(String id) { return entity != null ? Optional.of(EntityMapper.toDomain(entity)) : Optional.empty(); } + /** The run created with this Idempotency-Key, read without its task results. */ + public Optional findByClientKey(String clientKey) { + return em + .createQuery("SELECT r FROM TestRunEntity r WHERE r.clientKey = :clientKey", TestRunEntity.class) + .setParameter("clientKey", clientKey) + .getResultList() + .stream() + .findFirst() + .map(EntityMapper::toDomainSummary); + } + public List findAll() { return em .createQuery("SELECT r FROM TestRunEntity r ORDER BY r.createdAt DESC", TestRunEntity.class) diff --git a/kates/src/main/resources/db/migration/V29__test_run_client_key.sql b/kates/src/main/resources/db/migration/V29__test_run_client_key.sql new file mode 100644 index 000000000..f0d8553a1 --- /dev/null +++ b/kates/src/main/resources/db/migration/V29__test_run_client_key.sql @@ -0,0 +1,5 @@ +-- The caller's Idempotency-Key on POST /api/tests. Null on runs created +-- without one, and PostgreSQL lets a unique index hold any number of nulls. +ALTER TABLE test_runs ADD COLUMN client_key VARCHAR(128); + +CREATE UNIQUE INDEX idx_test_runs_client_key ON test_runs (client_key); diff --git a/kates/src/test/java/com/bmscomp/kates/api/HealthResourceTest.java b/kates/src/test/java/com/bmscomp/kates/api/HealthResourceTest.java index ec7456f93..e0ab1031e 100644 --- a/kates/src/test/java/com/bmscomp/kates/api/HealthResourceTest.java +++ b/kates/src/test/java/com/bmscomp/kates/api/HealthResourceTest.java @@ -47,6 +47,7 @@ void healthReturnsUpWithEngineAndPerTypeConfig() { .then() .statusCode(200) .body("status", is("UP")) + .body("capabilities", hasItem("clientKey")) .body("engine.activeBackend", is("native")) .body("engine.availableBackends", hasItems("native", "trogdor")) .body("kafka.status", is("UP")) diff --git a/kates/src/test/java/com/bmscomp/kates/api/TestResourceCreateTest.java b/kates/src/test/java/com/bmscomp/kates/api/TestResourceCreateTest.java index 5faf24e2b..afda556a8 100644 --- a/kates/src/test/java/com/bmscomp/kates/api/TestResourceCreateTest.java +++ b/kates/src/test/java/com/bmscomp/kates/api/TestResourceCreateTest.java @@ -11,6 +11,7 @@ import java.util.List; import java.util.Set; import jakarta.persistence.PersistenceException; +import jakarta.ws.rs.core.HttpHeaders; import jakarta.ws.rs.core.Response; import org.junit.jupiter.api.BeforeEach; @@ -45,6 +46,8 @@ class TestResourceCreateTest { private static final String NOT_STARTED = "The Kates API failed to start the test run — see server logs for details"; + private static final HttpHeaders NO_KEY = mock(HttpHeaders.class); + private final InMemoryEngine engine = new InMemoryEngine(); private final AuditService audit = mock(AuditService.class); private TestResource resource; @@ -59,7 +62,7 @@ void setUp() { void aRunTheKatesApiCannotSaveIsAFaultOfTheKatesApi() { doThrow(UNREACHABLE).when(engine.repository).save(any()); - assertFaultOfTheKatesApi(resource.createTest(InMemoryEngine.request())); + assertFaultOfTheKatesApi(resource.createTest(InMemoryEngine.request(), NO_KEY)); assertEquals(0, engine.orchestrator.activeTestCount(), "the run gave back its slot"); verifyNoInteractions(audit); } @@ -68,7 +71,7 @@ void aRunTheKatesApiCannotSaveIsAFaultOfTheKatesApi() { void aScenarioTheKatesApiCannotSaveIsAFaultOfTheKatesApi() { doThrow(UNREACHABLE).when(engine.repository).save(any()); - assertFaultOfTheKatesApi(resource.createTest(scenario())); + assertFaultOfTheKatesApi(resource.createTest(scenario(), NO_KEY)); assertEquals(0, engine.orchestrator.activeTestCount(), "the run gave back its slot"); verifyNoInteractions(audit); } @@ -78,7 +81,8 @@ void aBackendTheRequestNamesThatTheKatesApiDoesNotHaveIsTheCallersToChange() { CreateTestRequest request = InMemoryEngine.request(); request.setBackend("no-such-backend"); - assertBadRequest("Backend not found: 'no-such-backend'. Available: [fake]", resource.createTest(request)); + assertBadRequest( + "Backend not found: 'no-such-backend'. Available: [fake]", resource.createTest(request, NO_KEY)); assertEquals(0, engine.orchestrator.activeTestCount()); } @@ -87,7 +91,8 @@ void aBackendTheScenarioNamesThatTheKatesApiDoesNotHaveIsTheCallersToChange() { CreateTestRequest request = scenario(); request.getScenario().setBackend("no-such-backend"); - assertBadRequest("Backend not found: 'no-such-backend'. Available: [fake]", resource.createTest(request)); + assertBadRequest( + "Backend not found: 'no-such-backend'. Available: [fake]", resource.createTest(request, NO_KEY)); assertEquals(0, engine.orchestrator.activeTestCount()); } @@ -96,27 +101,28 @@ void aDefaultBackendTheKatesApiDoesNotHaveIsAFaultOfTheKatesApi() { // kates.engine.default-backend names a backend the Kates API does not // have, and the request names none, so it cannot be at fault. TestOrchestrator misconfigured = mock(TestOrchestrator.class); - when(misconfigured.executeTest(any())) + when(misconfigured.executeTest(any(), any())) .thenReturn(Result.failure(new UnknownBackendException("nativ", List.of("native", "trogdor")))); resource = new TestResource(misconfigured, engine.repository, null); resource.auditService = audit; - assertFaultOfTheKatesApi(resource.createTest(InMemoryEngine.request())); - assertFaultOfTheKatesApi(resource.createTest(scenario())); + assertFaultOfTheKatesApi(resource.createTest(InMemoryEngine.request(), NO_KEY)); + assertFaultOfTheKatesApi(resource.createTest(scenario(), NO_KEY)); verifyNoInteractions(audit); // A request that names it is the caller's to change, as with any // other backend the Kates API does not have. CreateTestRequest named = InMemoryEngine.request(); named.setBackend("nativ"); - assertBadRequest("Backend not found: 'nativ'. Available: [native, trogdor]", resource.createTest(named)); + assertBadRequest( + "Backend not found: 'nativ'. Available: [native, trogdor]", resource.createTest(named, NO_KEY)); } @Test void aFullEngineIsStillToBeRetried() { engine.fill(); - Response response = resource.createTest(InMemoryEngine.request()); + Response response = resource.createTest(InMemoryEngine.request(), NO_KEY); assertEquals(429, response.getStatus()); assertEquals("60", response.getHeaderString("Retry-After")); @@ -131,7 +137,7 @@ void aFieldTheRunCannotHonourIsStillRefusedByName() { CreateTestRequest request = InMemoryEngine.request(); request.getSpec().setEnableCrc(true); - Response response = resource.createTest(request); + Response response = resource.createTest(request, NO_KEY); assertEquals(400, response.getStatus()); ApiError error = (ApiError) response.getEntity(); @@ -215,7 +221,7 @@ void aRunThatStartedIsAcceptedWhenItsAuditFails() { .when(audit) .record(anyString(), anyString(), anyString(), anyString()); - Response response = resource.createTest(InMemoryEngine.request()); + Response response = resource.createTest(InMemoryEngine.request(), NO_KEY); assertEquals(202, response.getStatus()); assertTrue(engine.rows.containsKey(((TestRun) response.getEntity()).getId())); diff --git a/kates/src/test/java/com/bmscomp/kates/api/TestResourceTest.java b/kates/src/test/java/com/bmscomp/kates/api/TestResourceTest.java index 04d0f0592..1afc80776 100644 --- a/kates/src/test/java/com/bmscomp/kates/api/TestResourceTest.java +++ b/kates/src/test/java/com/bmscomp/kates/api/TestResourceTest.java @@ -5,17 +5,25 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.mockito.Mockito.*; +import java.util.ArrayList; import java.util.List; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import jakarta.inject.Inject; +import jakarta.persistence.EntityManager; +import io.quarkus.narayana.jta.QuarkusTransaction; import io.quarkus.test.InjectMock; import io.quarkus.test.junit.QuarkusTest; import io.quarkus.test.junit.TestProfile; +import io.quarkus.test.junit.mockito.InjectSpy; import org.eclipse.microprofile.rest.client.inject.RestClient; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.ValueSource; import org.mockito.ArgumentCaptor; import com.bmscomp.kates.SchedulerOffTestProfile; @@ -40,9 +48,12 @@ class TestResourceTest { @InjectMock TopicService topicService; - @Inject + @InjectSpy TestRunRepository repository; + @Inject + EntityManager em; + @BeforeEach void setUp() { doNothing().when(topicService).createTopic(anyString(), anyInt(), anyInt(), any()); @@ -586,6 +597,238 @@ private static java.util.Map scenario(String name, String phaseN List.of(java.util.Map.of("name", phaseName, "phaseType", "STEADY", "durationMs", 60_000)))); } + private static final String FLAT_LOAD = "{\"type\": \"LOAD\", \"backend\": \"trogdor\", \"spec\": {}}"; + + private static final String SCENARIO = "{\"type\": \"LOAD\", \"backend\": \"trogdor\", \"scenario\": {" + + "\"baseSpec\": {}, \"phases\": [{\"name\": \"steady\", \"phaseType\": \"STEADY\"," + + " \"durationMs\": 60000}]}}"; + + private static String newKey() { + return UUID.randomUUID().toString(); + } + + private static io.restassured.response.Response postWithKey(String key, String body) { + return given().contentType("application/json") + .header("Idempotency-Key", key) + .body(body) + .when() + .post("/api/tests"); + } + + private void refuseTasks() { + when(trogdorClient.createTask(any())).thenThrow(new IllegalStateException("no coordinator in tests")); + } + + private long handlerAuditRows(String runId) { + return QuarkusTransaction.requiringNew() + .call(() -> em.createQuery( + "SELECT COUNT(a) FROM AuditEventEntity a WHERE a.action = 'CREATE' AND a.target = :id", + Long.class) + .setParameter("id", runId) + .getSingleResult()); + } + + @Test + void aRepeatedKeyAnswersTheFirstRunWithoutAnotherHandlerAuditRow() { + refuseTasks(); + String key = newKey(); + + String id = postWithKey(key, FLAT_LOAD) + .then() + .statusCode(202) + .body("clientKey", is(key)) + .extract() + .path("id"); + postWithKey(key, FLAT_LOAD).then().statusCode(200).body("id", is(id)).body("clientKey", is(key)); + + assertEquals(1, handlerAuditRows(id)); + } + + @Test + void differentKeysStartTwoRuns() { + refuseTasks(); + + String first = postWithKey(newKey(), FLAT_LOAD) + .then() + .statusCode(202) + .extract() + .path("id"); + String second = postWithKey(newKey(), FLAT_LOAD) + .then() + .statusCode(202) + .extract() + .path("id"); + + org.junit.jupiter.api.Assertions.assertNotEquals(first, second); + } + + /** A loser past the lookup would reach the unique column and answer 500. */ + @Test + void twoThreadsWithOneKeyStartOneRun() throws Exception { + refuseTasks(); + String key = newKey(); + CountDownLatch start = new CountDownLatch(1); + List> posts = new ArrayList<>(); + for (int i = 0; i < 2; i++) { + posts.add(CompletableFuture.supplyAsync(() -> { + try { + start.await(); + } catch (InterruptedException e) { + throw new IllegalStateException(e); + } + return postWithKey(key, FLAT_LOAD); + })); + } + start.countDown(); + + var first = posts.get(0).get(); + var second = posts.get(1).get(); + + assertEquals(java.util.Set.of(200, 202), java.util.Set.of(first.statusCode(), second.statusCode())); + assertEquals(first.path("id"), second.path("id")); + given().queryParam("clientKey", key).when().get("/api/tests").then().body("total", is(1)); + } + + @Test + void aScenarioSentTwiceWithOneKeyStartsOneRun() { + refuseTasks(); + String key = newKey(); + + String id = postWithKey(key, SCENARIO) + .then() + .statusCode(202) + .body("clientKey", is(key)) + .extract() + .path("id"); + postWithKey(key, SCENARIO).then().statusCode(200).body("id", is(id)); + } + + @Test + void theListFindsARunByItsClientKey() { + refuseTasks(); + String key = newKey(); + String id = postWithKey(key, FLAT_LOAD).then().statusCode(202).extract().path("id"); + + given().queryParam("clientKey", key) + .when() + .get("/api/tests") + .then() + .statusCode(200) + .body("total", is(1)) + .body("items.id", org.hamcrest.Matchers.contains(id)) + .body("items[0].clientKey", is(key)); + given().queryParam("clientKey", newKey()) + .when() + .get("/api/tests") + .then() + .statusCode(200) + .body("total", is(0)) + .body("items", empty()); + } + + /** {@code save} mutates the managed entity, so a write that drops the key would null a column GET reads. */ + @ParameterizedTest + @ValueSource(strings = {FLAT_LOAD, SCENARIO}) + void theClientKeySurvivesEveryStatusWrite(String body) { + refuseTasks(); + String key = newKey(); + String id = postWithKey(key, body).then().statusCode(202).extract().path("id"); + awaitStatus(id, "FAILED"); + + TestRun whole = repository.findById(id).orElseThrow(); + repository.save(whole.withStatus(TestResult.TaskStatus.RUNNING)); + repository.save(whole.withStatus(TestResult.TaskStatus.DONE)); + given().when().get("/api/tests/" + id).then().body("status", is("DONE")).body("clientKey", is(key)); + + // The timeout reaper writes back a run read without its results. + repository.save(repository.findByClientKey(key).orElseThrow().withStatus(TestResult.TaskStatus.FAILED)); + given().when() + .get("/api/tests/" + id) + .then() + .body("status", is("FAILED")) + .body("clientKey", is(key)); + } + + /** RestAssured trims the header, and a blank one reaches the API empty. */ + @ParameterizedTest + @ValueSource(strings = {" ", "tab\tinside"}) + void aBlankOrControlCharacterKeyIsRefusedByName(String key) { + verifyRefused(key); + } + + @Test + void aKeyOver128CharactersIsRefusedByName() { + verifyRefused("k".repeat(129)); + } + + private void verifyRefused(String key) { + postWithKey(key, FLAT_LOAD) + .then() + .statusCode(400) + .body("error", is("Validation Failed")) + .body("fieldErrors", hasKey("Idempotency-Key")); + verifyNoInteractions(trogdorClient); + } + + @ParameterizedTest + @ValueSource(strings = {FLAT_LOAD, SCENARIO}) + void aFailedSaveLeavesTheKeyForTheRetry(String body) { + refuseTasks(); + String key = newKey(); + doThrow(new IllegalStateException("database blip")) + .doCallRealMethod() + .when(repository) + .save(argThat(run -> key.equals(run.getClientKey()))); + + postWithKey(key, body).then().statusCode(500); + postWithKey(key, body).then().statusCode(202).body("clientKey", is(key)); + } + + @Test + void anSlaBesideAScenarioIsRefused() { + given().contentType("application/json") + .body(SCENARIO.substring(0, SCENARIO.length() - 1) + ", \"sla\": {\"maxP99LatencyMs\": 1}}") + .when() + .post("/api/tests") + .then() + .statusCode(400) + .body("fieldErrors", hasKey("sla")); + verifyNoInteractions(trogdorClient); + } + + @Test + void aFlatRequestsSlaFailsTheReportWhenBreached() { + refuseTasks(); + String id = given().contentType("application/json") + .body("{\"type\": \"LOAD\", \"backend\": \"trogdor\", \"spec\": {}," + + " \"sla\": {\"maxP99LatencyMs\": 1}}") + .when() + .post("/api/tests") + .then() + .statusCode(202) + .body("sla.maxP99LatencyMs", is(1.0f)) + .extract() + .path("id"); + awaitStatus(id, "FAILED"); + + var measured = new TestResult() + .withTaskId("t-sla") + .withStatus(TestResult.TaskStatus.DONE) + .withRecordsSent(1000) + .withP99LatencyMs(50); + repository.save(repository + .findById(id) + .orElseThrow() + .withResults(List.of(measured)) + .withStatus(TestResult.TaskStatus.DONE)); + + given().when() + .get("/api/tests/" + id + "/report") + .then() + .statusCode(200) + .body("overallSlaVerdict.passed", is(false)); + } + private static void awaitStatus(String id, String status) { long deadline = System.currentTimeMillis() + 10_000; while (System.currentTimeMillis() < deadline) { diff --git a/kates/src/test/java/com/bmscomp/kates/domain/WireSchemaTest.java b/kates/src/test/java/com/bmscomp/kates/domain/WireSchemaTest.java new file mode 100644 index 000000000..e045aa29d --- /dev/null +++ b/kates/src/test/java/com/bmscomp/kates/domain/WireSchemaTest.java @@ -0,0 +1,64 @@ +package com.bmscomp.kates.domain; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.Set; +import java.util.TreeSet; +import java.util.stream.Collectors; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.introspect.BeanPropertyDefinition; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +/** + * The API's request classes against the schema the operator's CRD is generated + * from: a field added to one of them fails here until the schema has it, since + * the apiserver prunes a key the CRD does not know. + */ +class WireSchemaTest { + + private static final Path SCHEMA = Path.of("..", "cli", "schemas", "test-request.v1.schema.json"); + + /** Keys kates lint and the LSP accept and the API does not bind; crdgen drops the same list. */ + private static final Set FILE_ONLY = Set.of( + "slaDefinition.p95LatencyMs", + "slaDefinition.p99LatencyMs", + "slaDefinition.avgLatencyMs", + "slaDefinition.maxErrorRatePercent", + "slaDefinition.maxRecoveryTimeMs", + "slaDefinition.maxThroughputDegradationPercent", + "slaDefinition.maxIsrRecoveryTimeMs", + "slaDefinition.maxConsumerLag", + "scenarioPhase.type", + "scenarioPhase.specOverrides"); + + @ParameterizedTest + @CsvSource({ + "com.bmscomp.kates.domain.TestSpec, wireTestSpec", + "com.bmscomp.kates.domain.SlaDefinition, slaDefinition", + "com.bmscomp.kates.domain.TestScenario, multiPhaseScenario", + "com.bmscomp.kates.domain.ScenarioPhase, scenarioPhase" + }) + void jacksonBindsExactlyWhatTheSchemaDefines(Class type, String def) throws IOException { + ObjectMapper mapper = new ObjectMapper(); + Set bound = + mapper.getDeserializationConfig().introspect(mapper.constructType(type)).findProperties().stream() + .filter(BeanPropertyDefinition::couldDeserialize) + .map(BeanPropertyDefinition::getName) + .collect(Collectors.toCollection(TreeSet::new)); + + Set schema = new TreeSet<>(); + JsonNode properties = mapper.readTree(SCHEMA.toFile()).at("/$defs/" + def + "/properties"); + properties.fieldNames().forEachRemaining(name -> { + if (!FILE_ONLY.contains(def + "." + name)) { + schema.add(name); + } + }); + + assertEquals(schema, bound); + } +} diff --git a/kates/src/test/java/com/bmscomp/kates/it/IdempotentCreateIT.java b/kates/src/test/java/com/bmscomp/kates/it/IdempotentCreateIT.java new file mode 100644 index 000000000..fb0f9433d --- /dev/null +++ b/kates/src/test/java/com/bmscomp/kates/it/IdempotentCreateIT.java @@ -0,0 +1,140 @@ +package com.bmscomp.kates.it; + +import static io.restassured.RestAssured.given; +import static org.hamcrest.Matchers.is; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doThrow; + +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import jakarta.inject.Inject; +import jakarta.persistence.EntityManager; + +import io.quarkus.narayana.jta.QuarkusTransaction; +import io.quarkus.test.InjectMock; +import io.quarkus.test.common.QuarkusTestResource; +import io.quarkus.test.junit.QuarkusTest; +import io.quarkus.test.junit.TestProfile; +import io.restassured.http.ContentType; +import org.junit.jupiter.api.Test; + +import com.bmscomp.kates.domain.CreateTestRequest; +import com.bmscomp.kates.domain.TestRun; +import com.bmscomp.kates.domain.TestType; +import com.bmscomp.kates.engine.ConcurrencyLimitException; +import com.bmscomp.kates.engine.TestOrchestrator; +import com.bmscomp.kates.service.TestRunRepository; +import com.bmscomp.kates.service.TopicService; +import com.bmscomp.kates.util.Result; + +/** + * The unique index V29 builds, on the path only a second pod or a replay after + * a restart can reach: past the resource's lookup and its per-key lock. + * + *

H2 cannot hold this: the insert's failure surfaces at commit on Postgres, + * as a different exception through the transaction interceptor. + */ +@QuarkusTest +@TestProfile(NoSchedulersTestProfile.class) +@QuarkusTestResource(value = PostgresTestResource.class, restrictToAnnotatedClass = true) +class IdempotentCreateIT { + + @InjectMock + TopicService topicService; + + @Inject + TestOrchestrator orchestrator; + + @Inject + TestRunRepository repository; + + @Inject + EntityManager em; + + /** + * EXCLUSIVE mode refuses writers and lets readers through, so both creates + * pass every read and queue on the same insert, which is the race itself. + */ + @Test + void theLoserOfTheIndexRaceGivesItsPermitBack() { + doThrow(new IllegalStateException("no broker here")) + .when(topicService) + .createTopic(anyString(), anyInt(), anyInt(), any()); + String key = UUID.randomUUID().toString(); + CreateTestRequest request = ItSupport.createRequest(TestType.LOAD, ItSupport.singleBrokerSpec(null)); + // The bean's first use reads the database, which needs this thread's request context. + assertEquals(0, orchestrator.activeTestCount()); + + List> results = QuarkusTransaction.requiringNew() + .call(() -> { + em.createNativeQuery("LOCK TABLE test_runs IN EXCLUSIVE MODE") + .executeUpdate(); + var first = CompletableFuture.supplyAsync(() -> orchestrator.executeTest(request, key)); + var second = CompletableFuture.supplyAsync(() -> orchestrator.executeTest(request, key)); + assertTrue(ItSupport.waitUntil(() -> waitingWriters() == 2), "both creates reached the insert"); + return List.of(first, second); + }) + .stream() + .map(CompletableFuture::join) + .toList(); + + assertEquals(1, results.stream().filter(Result::isSuccess).count()); + Exception lost = results.stream() + .filter(Result::isFailure) + .findFirst() + .orElseThrow() + .asFailure() + .orElseThrow(); + assertTrue(!(lost instanceof ConcurrencyLimitException), lost.toString()); + + assertTrue( + ItSupport.waitUntil(() -> orchestrator.activeTestCount() == 0), + "permits still held: " + orchestrator.activeTestCount()); + + // The key stays with the winner; the loser's failed row holds none. + String id = results.stream() + .filter(Result::isSuccess) + .findFirst() + .orElseThrow() + .asSuccess() + .orElseThrow() + .getId(); + assertEquals(id, repository.findByClientKey(key).orElseThrow().getId()); + + // The loser's replay finds the run, and a new key still starts one. + post(key).then().statusCode(200); + post(UUID.randomUUID().toString()).then().statusCode(202); + } + + /** Postgres refuses NUL in a text parameter, which H2 does not. */ + @Test + void aClientKeyNoRunCanHoldIsFoundOnNoRun() { + given().queryParam("clientKey", "a" + (char) 0 + "b") + .when() + .get("/api/tests") + .then() + .statusCode(200) + .body("total", is(0)); + } + + private long waitingWriters() { + return ((Number) em.createNativeQuery("SELECT COUNT(*) FROM pg_locks l WHERE l.relation = 'test_runs'::regclass" + + " AND NOT l.granted") + .getSingleResult()) + .longValue(); + } + + private static io.restassured.response.Response post(String key) { + return given().contentType(ContentType.JSON) + .header("Idempotency-Key", key) + .body("{\"type\":\"LOAD\",\"backend\":\"native\",\"spec\":{\"numRecords\":25,\"partitions\":1," + + "\"replicationFactor\":1,\"minInsyncReplicas\":1}}") + .when() + .post("/api/tests"); + } +} diff --git a/schemas/test-request.v1.schema.json b/schemas/test-request.v1.schema.json index 759d347fd..e5f9741a4 100644 --- a/schemas/test-request.v1.schema.json +++ b/schemas/test-request.v1.schema.json @@ -9,6 +9,7 @@ { "required": ["type"] }, { "required": ["scenario"] } ], + "not": { "required": ["sla", "scenario"] }, "properties": { "type": { "description": "Test type for a single-phase run (or fallback if scenario.type is omitted).", @@ -22,6 +23,10 @@ "description": "Wire-format TestSpec parameters (uses numRecords, numProducers, recordSize, durationMs, throughput).", "$ref": "#/$defs/wireTestSpec" }, + "sla": { + "description": "Pass/fail criteria for a run with no scenario; a scenario carries its own under scenario.sla.", + "$ref": "#/$defs/slaDefinition" + }, "scenario": { "description": "Multi-phase scenario envelope with baseSpec, ordered execution phases, SLA definition, and metadata labels.", "$ref": "#/$defs/multiPhaseScenario" From 545658ae08153edc221bd97f4452317de7801a97 Mon Sep 17 00:00:00 2001 From: Said Boudjelda Date: Thu, 5 Nov 2026 18:23:06 +0100 Subject: [PATCH 2/2] fix(api): refuse a replication factor that would wrap when cast to short CodeQL flagged the cast in TopicService.createTopic once this branch's request path reached it: an int too large for a short silently wrapped into a different replication factor. Out-of-range values now fail with IllegalArgumentException, which @Retry already aborts on, and POST /topics answers 400 rather than a 500. -1 still means the broker default. Co-Authored-By: Claude Opus 5.5 --- .../bmscomp/kates/api/KafkaClientResource.java | 4 ++++ .../bmscomp/kates/service/TopicService.java | 4 ++++ .../kates/service/TopicServiceTest.java | 18 ++++++++++++++++++ 3 files changed, 26 insertions(+) diff --git a/kates/src/main/java/com/bmscomp/kates/api/KafkaClientResource.java b/kates/src/main/java/com/bmscomp/kates/api/KafkaClientResource.java index e0d2426d8..b1a1977ca 100644 --- a/kates/src/main/java/com/bmscomp/kates/api/KafkaClientResource.java +++ b/kates/src/main/java/com/bmscomp/kates/api/KafkaClientResource.java @@ -217,6 +217,10 @@ public Response createTopic(CreateTopicRequest request) { topicService.createTopic(request.name(), partitions, rf, request.configs()); var detail = topicService.describeCreatedTopic(request.name()); return Response.status(201).entity(detail).build(); + } catch (IllegalArgumentException e) { + return Response.status(400) + .entity(ApiError.of(400, "Bad Request", e.getMessage())) + .build(); } catch (Exception e) { return Response.serverError() .entity(ApiError.of(500, "Kafka Error", e.getMessage())) diff --git a/kates/src/main/java/com/bmscomp/kates/service/TopicService.java b/kates/src/main/java/com/bmscomp/kates/service/TopicService.java index 950ca4943..d29f45150 100644 --- a/kates/src/main/java/com/bmscomp/kates/service/TopicService.java +++ b/kates/src/main/java/com/bmscomp/kates/service/TopicService.java @@ -53,6 +53,10 @@ public void evictCache() { abortOn = {org.apache.kafka.common.errors.TopicExistsException.class, IllegalArgumentException.class}) @Timeout(35_000) public void createTopic(String name, int partitions, int replicationFactor, Map configs) { + // -1 asks the broker for its default; anything past a short would wrap silently in the cast below + if (replicationFactor < -1 || replicationFactor > Short.MAX_VALUE) { + throw new IllegalArgumentException("replicationFactor out of range: " + replicationFactor); + } AdminClient client = adminService.getClient(); try { NewTopic newTopic = new NewTopic(name, partitions, (short) replicationFactor); diff --git a/kates/src/test/java/com/bmscomp/kates/service/TopicServiceTest.java b/kates/src/test/java/com/bmscomp/kates/service/TopicServiceTest.java index ff909dba0..262900d09 100644 --- a/kates/src/test/java/com/bmscomp/kates/service/TopicServiceTest.java +++ b/kates/src/test/java/com/bmscomp/kates/service/TopicServiceTest.java @@ -65,6 +65,24 @@ void createTopicWithNullConfigsSucceeds() { assertDoesNotThrow(() -> topicService.createTopic("test-topic", 1, 1, null)); } + @Test + void createTopicRefusesAReplicationFactorThatWouldWrap() { + assertThrows( + IllegalArgumentException.class, + () -> topicService.createTopic("test-topic", 1, Short.MAX_VALUE + 1, null)); + assertThrows(IllegalArgumentException.class, () -> topicService.createTopic("test-topic", 1, -2, null)); + verify(mockClient, never()).createTopics(any()); + } + + @Test + void createTopicKeepsTheBrokerDefaultReplicationFactor() { + CreateTopicsResult result = mock(CreateTopicsResult.class); + when(result.all()).thenReturn(KafkaFuture.completedFuture(null)); + when(mockClient.createTopics(any())).thenReturn(result); + + assertDoesNotThrow(() -> topicService.createTopic("test-topic", 1, -1, null)); + } + @Test void deleteTopicSucceeds() { DeleteTopicsResult result = mock(DeleteTopicsResult.class);