KAFKA-21074: Kafka protocol fault proxy clients fixtures - #23438
Conversation
A client-agnostic Kafka wire-protocol fault-injection proxy for integration tests: sit it in front of an EmbeddedKafkaCluster and point clients at it to inject error codes, drop connections, or blackhole a client by id. Decodes with Kafka's own protocol classes so it is correct across wire versions. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
github-actions
Bot
added
streams
tests
| final byte[] frame = new byte[size]; | ||
| in.readFully(frame); | ||
| return frame; | ||
| } catch (final EOFException eof) { |
| } catch (final Exception ignored) { | ||
| // closing | ||
| } | ||
| threadPool.shutdownNow(); |
There was a problem hiding this comment.
Should we await termination here?
| final Socket client = serverSocket.accept(); | ||
| final Socket broker = new Socket(targetHost, targetPort); | ||
| final Connection conn = new Connection(); | ||
| threadPool.submit(() -> pumpRequests(client, broker, conn)); |
There was a problem hiding this comment.
How do we close these connections when the test fails? They are only owned by the pump threads
| }, 60_000L, "counts did not converge to " + numRecords); | ||
|
|
||
| // The fault actually fired (guards against a hollow pass). | ||
| assertTrue(fence.timesTriggered() >= 1, "END_TXN fence never fired"); |
There was a problem hiding this comment.
I ran this test against the PR branch and it fails deterministically with AssertionFailedError: END_TXN fence never fired ==> expected: <true> but was: <false>. The state store sum reaches numRecords before the first commit/END_TXN ever happens (Streams updates the local store synchronously as records are processed, independent of the commit boundary), so this assertion fires before the fence has a chance to. The shutdown-time commit that actually gets fenced happens too late.
| for (final FaultRule rule : rules) { | ||
| // Gate on apiKey AND clientId before shouldFire(), so a client-scoped rule only counts (and | ||
| // fires on) matching requests — a fetch fault scoped to "restore" ignores the main consumer. | ||
| if (rule.apiKey() == apiKey && rule.matchesClient(clientId) && rule.shouldFire() && chosen == null) { |
There was a problem hiding this comment.
Since shouldFire() is called on every matching rule (not just the one that ends up chosen), two unscoped rules on the same API both get their match counted even though only the first one's action actually runs. E.g. injectError(FETCH, X).once() and disconnectOn(FETCH).once() registered together: on the first matching FETCH, both report timesTriggered()==1, but only the first rule fires. That silently breaks the timesTriggered() >= 1 oracle the example tests use.
Yep — it lives in clients/src/testFixtures, so any module that adds testImplementation testFixtures(project(':clients')) can use it (already ~37 do). Streams is just the first user here. |
|
@chia7712 did you want to make another pass before I merge? |
|
Merged #23438 into trunk |
chia7712
left a comment
chia7712
left a comment
There was a problem hiding this comment.
@bbejeckthanks for this patch. The idea is cool. I will take a look at the clients module to see if there are existing tests that can leverage this mechanism.
| private volatile int proxyPort; | ||
|
|
||
| private KafkaProtocolFaultProxy(final String targetBootstrap) { | ||
| final String hostPort = targetBootstrap.split(",")[0].trim(); |
There was a problem hiding this comment.
Does it assume there is only one server? If so, we should throw an exception if the passed targetBootstrap contains multiple servers.
There was a problem hiding this comment.
No I don't think so becuase in integration tests we specify any number of brokers (between 1 and 3 from my experience) I'll drill into this a little more and file a follow-up PR if need be.
3 participants
A lightweight Kafka wire-protocol fault-injection proxy for integration
tests. Sit it in front of an EmbeddedKafkaCluster, point any client’s
bootstrap.servers at it, and it can inject error codes, drop
connections, or blackhole a client — decoding/encoding with Kafka’s own
protocol classes so it is correct across every wire version (including
flexible/tagged-field ones), no hand-rolled byte offsets.
Placed in
clients/src/testFixturesso it is reusable from any module.This is internal test infrastructure only so it does not require a KIP.
Fault primitives, each armed with a fluent, deterministic trigger:
injectError(apiKey, Errors)— stamp an error code onto the matchingresponse, re-serialized at the correct wire version. Supported APIs:
END_TXN,INIT_PRODUCER_ID,ADD_OFFSETS_TO_TXN,TXN_OFFSET_COMMIT,PRODUCE,FETCH(extensible by registering aper-API setter).
disconnectOn(apiKey)— drop the connection when thematching response would return (models the EOS "commit gap"). Works
on any API.
delayOn(apiKey, Duration)— hold back the matchingresponse to model a slow broker, isolated to that one connection.
blackholeClient(clientIdSubstring)— drop all requests from a targetedclient before the broker sees them, so the broker evicts it by
session timeout (a reversible, ungraceful one-node partition). Shared
trigger DSL (
Occurrence):once(),onCall,times,everyTime(),withProbability(p). The deterministic triggers are safe forassertions; probability is chaos-mode only. The proxy never closes
sockets unless a
disconnectOn(...)rule fires, so it is not itself asource of flakiness. Rules can be armed/disarmed live from the test
thread and each returns a handle exposing match/fire counts.
Reviewers: Chia-Ping Tsai chia7712@gmail.com, Lucas Brutschy
lbrutschy@confluent.io