KAFKA-9608: Transaction Event Simulation Test #8725
Conversation
Random drop message support add request completion temporal committed offsets
abbccdda
force-pushed
the
txn_simulation_test
branch
4 times, most recently
from
2a3261c
to
61414e9
Compare
abbccdda
force-pushed
the
txn_simulation_test
branch
from
61414e9
to
4c352e0
Compare
|
retest this please |
...rc/test/java/org/apache/kafka/clients/producer/internals/TransactionEventSimulationTest.java
Show resolved
Hide resolved
...rc/test/java/org/apache/kafka/clients/producer/internals/TransactionEventSimulationTest.java
Show resolved
Hide resolved
...rc/test/java/org/apache/kafka/clients/producer/internals/TransactionEventSimulationTest.java
Show resolved
Hide resolved
.../test/java/org/apache/kafka/clients/producer/internals/TransactionSimulationCoordinator.java
Show resolved
Hide resolved
| void runOnce(boolean dropMessage) { | ||
| Queue<ClientRequest> incomingRequests = networkClient.requests(); | ||
|
|
||
| final boolean faultInject = faultInjectRandom.nextBoolean(); |
There was a problem hiding this comment.
Wondering if it would be more useful if we can control the faults more explicitly. For example, we could add a hook to make the coordinator temporarily unavailable and to restore it later.
| pendingOffsets.put(key, partition.committedOffset()); | ||
| errors.put(key, Errors.NONE); | ||
| } else { | ||
| errors.put(key, Errors.UNKNOWN_TOPIC_OR_PARTITION); |
There was a problem hiding this comment.
In this case, the client has send TxnOffsetCommit before the partition has been added to the transaction, which is an illegal state transition. Could we throw an assertion error or something directly to cause the test to fail?
| ); | ||
| } | ||
|
|
||
| private AbstractResponse handleProduce(ProduceRequest request, |
There was a problem hiding this comment.
To make this really interesting, we would need to add some sequence number bookkeeping. Really its the sequence/epoch bookkeeping which makes the implementation so complex.
| // Trigger KIP-360 path. | ||
| errors.put(topicPartition, new PartitionResponse(Errors.UNKNOWN_PRODUCER_ID)); | ||
| } else { | ||
| List<Record> sentRecords = pendingPartitionData.getOrDefault(topicPartition, new ArrayList<>()); |
There was a problem hiding this comment.
Similar to the offset commit path, it would be useful to validate here that each partition that was written to was first added to the transaction properly.
There was a problem hiding this comment.
made an attempt on that, but since we don't have a retry mechanism yet, it seems a bit hard to assume AddPartition to be successful before produce.
| return new ProduceResponse(errors, throttleTimeMs); | ||
| } | ||
|
|
||
| private EndTxnResponse handleEndTxn(EndTxnRequest request, final boolean faultInject) { |
There was a problem hiding this comment.
Another class of failure that we can simulate is when a request reaches the broker and gets handled, but the connection is lost before the response is sent.
There was a problem hiding this comment.
Thought we do have that coverage on L126:
networkClient.respond(response, disconnect);
Start the basic template for transaction event simulation.
Committer Checklist (excluded from commit message)