[LautaroPetaccio]

Eighth in the AsyncAPI white-box stack, on top of #1874. With the service's Kafka client instrumented, EvoMaster sees what the service consumes and emits, and a settle window can end when the service is done with the message, rather than when its coverage stays stable.

Everything new is behind two experimental options, both off by default: --instrumentMR_KAFKA and --asyncApiCompletion HANDLER_RETURN.

What it adds

Piece	What it does
KafkaConsumerClassReplacement, KafkaConsumerInterfaceClassReplacement	poll(Duration) on KafkaConsumer and on the Consumer interface.
KafkaConsumerLegacyPollClassReplacement, KafkaConsumerInterfaceLegacyPollClassReplacement	poll(long), in classes of their own (see below).
KafkaProducerClassReplacement, KafkaProducerInterfaceClassReplacement	send(record) and send(record, callback) on KafkaProducer and on the Producer interface.
KafkaReplacementHelper	What they share: the call to the original, reading records by reflection (kafka-clients is a test dependency only), the driver check and the re-entrancy guard.
ReplacementCategory.KAFKA, --instrumentMR_KAFKA	The new category, experimental and off by default, added to methodReplacementCategories().
EmittedMessage / EmittedMessageDto	A message the service handed to the producer: address, payload as text (cut at 4096 characters, never between the two halves of a surrogate pair), whether it was cut, and the payload's class. At most 100 per action.
AdditionalInfo / AdditionalInfoDto	emittedMessages per action, copied in EMController.computeTestResults.
AsyncApiCallResult.getEmittedMessages()	AsyncApiFitness keeps on each action's result what the service emitted while it was handled, in white-box mode only: address, payload cut at 1024 characters, whether it was cut, and the payload's class, as results are kept for every action in the archive. This is the surface the next PR turns into targets and assertions.
AsyncMessageTracker	Per test, per thread, the batches a poll returned, and whether the thread has polled again since. Reset with ExecutionTracer; a batch read into a test that is already over is dropped.
MessageHandlingState	What the tracker answers for an address in the current action: received and unhandled.
ExecutionTracer.setExecutingDriverAction	A per-thread flag EMController sets around executeAsyncApiAction; calls made on that thread while it is set are not recorded.
HANDLER_RETURN	Appended to --asyncApiCompletion, AsyncApiActionDto.MessageCompletionStrategy and AsyncApiReplyDto.SettleEndReason. AsyncApiSettle ends the window once something was polled from the published address in this action and all of it is handled, capped by --asyncApiSettleMs.
SutController.getMessageHandlingState / Command.MESSAGE_HANDLING_STATE	Returns null by default. The embedded controller reads the tracker directly, and returns null without the agent; the external one asks the agent, the way OBJECTIVE_UPDATE_COUNT works, null included when the agent fails to read it. A state that cannot be read means the rest of the window.

Things that are less obvious than they look

• Both the class and the interface. Replacements match the class a call is compiled against. NCS declares KafkaConsumer/KafkaProducer fields; Spring Kafka's listener container polls a Consumer and KafkaTemplate sends through a Producer. Library code is instrumented too, so the interface replacements fire inside Spring.
• poll(long) is in classes of its own. initMethods resolves every replacement of a class the first time one is used, and a missing method breaks all of them. poll(long) is gone from kafka-clients 4.0 (checked against the 4.0.0 jar with javap; poll(Duration) and both send signatures are still there). Next to poll(Duration), it would break every poll on 4.x.
• One record, one emitted message. Spring's CloseSafeProducer sends every record twice through Producer.send (once from KafkaTemplate, once to its delegate), and KafkaProducer.send(record) calls send(record, null) on itself. A send made from inside another replaced send on the same thread is not recorded again. KafkaClientsTest checks both through the instrumenting class loader, and both fail without the guard. The price: a send that a callback or a custom producer makes from inside another send, on the same thread, is not recorded either.
• The driver's own calls. A driver class loaded once the agent is in place and the replacement categories are set gets the replacements too, and its package can be the service's. With the flag off, NCS's driver loaded that way has its publishes to the request topics recorded as messages the service emitted, and the new E2E assertion catches it. The flag only covers the thread running executeAsyncApiAction, while it runs: calls a driver makes on threads it starts itself, such as a consumer it keeps polling in the background, are recorded as the service's, and docs/write_driver.md says so. Being per thread, it leaves the service's consumer threads recorded while the driver publishes.
• Only text payloads are read. A String is recorded as given, and bytes as UTF-8, decoding no more of them than is kept. Any other object, boxed primitives included, is kept as its class name only: its bytes on the wire are up to the serializer (a Long is 8 binary bytes with LongSerializer), and calling toString would run SUT code from inside the replacement. The text is what goes on the wire only with StringSerializer or ByteArraySerializer: spring-kafka's JsonSerializer, for one, writes a String quoted.
• What "handled" means. A batch counts as handled when the thread that polled it polls again with a timeout, which is what a consumer loop does once its handler has returned. It is marked before the call, which can block for the whole poll timeout.

The completion is a heuristic, and these are its blind spots

• Work handed off to another thread (an executor, an async listener) looks done as soon as it is handed off: the receiving thread polls again at once. The Shop E2E has exactly that, an invoice rendered on an executor, and does not assert it is covered under this completion.
• A poll with a zero timeout does not count as the handler having returned, because Spring Kafka polls that way to keep the consumer alive while it retries a failed batch. A loop that only ever polls with a zero timeout never looks done, and every wait goes to the deadline.
• A failed record that an error handler seeks back to reads as handled in the moment between the failed attempt and the next poll's return. When it comes back it is unhandled again.
• What was polled in the current action counts, so a message left over from an earlier action and only consumed now counts too.
• A thread that closes its consumer without polling again leaves its last batch unhandled, and the wait goes to the deadline.

What this changes outside AsyncAPI

• ReplacementList.getReplacements asserted that a constructor lookup finds at most one replacement class per target. It asserts at most one class that replaces constructors instead, since the two KafkaConsumer classes replace none. MethodReplacementMethodVisitor takes the consume method from the class of the chosen replacement instead of the first candidate. For every existing replacement there is exactly one class per target, so nothing changes for them. Happy to split that into its own PR if you would rather.
• ExecutionTracer.reset() also resets AsyncMessageTracker, after releasing its own lock. ExecutionTracer.getActionIndex() is new.
• client-java/instrumentation/pom.xml: kafka-clients in test scope (it already came through spring-kafka; the version is managed in the root pom).
• docs/write_driver.md: one paragraph on what --instrumentMR_KAFKA records, that the driver's calls on the thread running executeAsyncApiAction are not taken for the service's, and that calls on threads it starts itself can be. docs/options.md regenerated.

Not here

• Targets and assertions on the emitted messages: the next PR.
• The bytes an object payload is serialized to. They are only produced inside KafkaProducer by the configured serializer, so a payload sent as an object is recorded as its class name.
• Other Kafka clients, such as Kafka Streams or reactor-kafka, and other transports.

Testing

• AsyncMessageTrackerTest (new): 13 tests. Handled on the same thread's next waiting poll, not on a zero-timeout poll nor on another thread's; batches per thread, per address and per action; a message received again is unhandled again; a batch read into a test that is over is dropped; reset.
• KafkaConsumerInterfaceClassReplacementTest (8), KafkaConsumerInterfaceLegacyPollClassReplacementTest (3), against kafka-clients' MockConsumer: records reach the caller, tracker state, zero timeout, driver flag, exceptions go through unchanged, a record polled again is unhandled again.
• KafkaConsumerClassReplacementTest (2), KafkaConsumerLegacyPollClassReplacementTest (2), against a real KafkaConsumer with no broker: an unsubscribed consumer fails before connecting, which shows the original is reached and the previous batch is handled.
• KafkaProducerInterfaceClassReplacementTest (14) against MockProducer, KafkaProducerClassReplacementTest (3) against a real KafkaProducer with no broker: what is recorded, callbacks, bytes, only a prefix of long bytes decoded, boxed primitives and object payloads never turned into text, truncation, the per-action bound, driver flag, a send that throws, a wrapping producer recorded once.
• KafkaClientsTest (new, 6 tests): the replacements applied by the instrumenting class loader to SUT code declaring the classes and the interfaces, poll(long), KafkaProducer calling itself and KafkaTemplate over CloseSafeProducer each recorded once, nothing tracked without the category.
• EmittedMessageTest (6: the cut, never inside a surrogate pair), MessageHandlingStateTest (2), ExecutionTracerTest (+2: flag per thread, recording), ServerControllerTest (+2: the new command, and a question the agent fails to answer gets null), ReplacementListTest (+1).
• All of client-java/instrumentation: 838 tests, 0 failures, 1 skipped (TestabilityExcInstrumentedTest, already disabled).
• AsyncApiSettleTest (+11): handled at once, after a while, nothing received, never handled, state or address unreadable, a state that fails to be read, the address asked is the published one, other completions never read it, interrupt. EMControllerTest (+6): the driver runs flagged and the flag is cleared after a failure, a wait ended by a consumer thread polling again through the real tracker, the deadline when nothing is consumed, the whole window without the agent, emitted messages in the test results.
• All of client-java/controller: 1514 tests, 0 failures, 2 skipped. RedisHeuristicsCalculatorTest.testKeysPatternNoMatch is flaky, and does not touch anything changed here.
• Core: EMConfigTest (+2: HANDLER_RETURN needs a settle window and the Kafka instrumentation; the KAFKA category only when asked for), AsyncApiFitnessTest (+4: HANDLER_RETURN sent, emitted messages kept per action and through a copy, fewer entries than actions, none in black-box), AsyncApiCallResultTest (new, 5: what a result keeps of an emitted message, a payload cut at 1024 characters and flagged, never inside a surrogate pair, the type of a payload with no text). AsyncApi*Test, ConfigToMarkdownTest and EMConfigTest: 296 tests, 0 failures.
• NcsKafkaEMTest (+1, testRunEMWithHandlerReturn): a 5 s window ends on HANDLER_RETURN every time and never on the deadline, every reply still arrives, bessj and checkTriangle still reach their results, and every emitted message recorded is on a reply topic. The whole class: 6 tests, 0 failures.
• ShopKafkaHandlerReturnEMTest (new, in spring-asyncapi-kafka-listener): the Shop service's three @KafkaListeners, whose containers poll through Consumer. Every operation has waits that end on HANDLER_RETURN, none goes to the 5 s deadline, ExpressDispatch and UrgentEscalation are still reached by taint (the listener compares the strings before it returns), and nothing is recorded as emitted, as the service publishes nothing (it has no KafkaTemplate or producer). The invoice rendered on an executor is not asserted on. It is a class of its own because the Kafka replacements are only applied to classes loaded after they are asked for, and the service, Spring's containers included, is started once in initClass; the module runs each class in a JVM of its own, so ShopKafkaEMTest is unchanged. The whole module: 3 tests, 0 failures.