Repository: kroxylicious/kroxylicious
Pull request #516 (2023-08-04 02:35:54)

Prevent short-circuit response to zero-acks Produce Request
### Type of change

- Enhancement / new feature

### Description

1. Moves the detection of zero-acks Produce Requests earlier in the chain to the `KafkaRequestDecoder`, so we know about it as soon as possible.
2. Drop short-circuit responses instead of forwarding if the Filter attempts to short-circuit respond to a zero-ack Produce Request.

### Additional Context

This PR was sparked by #507 because we need to know earlier whether a particular Request expects a Response according to the Kafka Protocol. In conversation about this with @SamBarker I mentioned that Filters could also break the new message re-ordering logic if they short-circuit responded in reaction to a zero-ack Produce Request, even if we avoid adding it's correlationId to the queue. We thought it would be worth preventing Filters from doing this.

### Checklist

- [ ] Write tests
- [ ] Make sure all tests pass
- [ ] Review performance test results. Ensure that any degradations to performance numbers are understood and justified.
- [ ] Make sure all Sonar-Lint warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [ ] Reference relevant issue(s) and close them after merging
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=516)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [11 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL)

[![83.1%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '83.1%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list) [83.1% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list)


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=516)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [11 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL)

[![83.1%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '83.1%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list) [83.1% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list)


robobario: Sonar registering a couple of valid smells about complexity of FilterHandler methods but I won't address them here
robobario: Wonder if we have an Integration Test covering zero-ack produce, could be worth checking the basic forwarding and sendRequest cases since we have to specialise
sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=516)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL)

[![83.8%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '83.8%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list) [83.8% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list)


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=516)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL)

[![83.8%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '83.8%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list) [83.8% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list)


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=516)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL)

[![83.8%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '83.8%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list) [83.8% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list)


robobario: Added integration tests for zero-ack produce requests, the code feels special enough to warrant a test checking they make it to the upstream broker.
sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=516)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL) [7 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=516&resolved=false&types=CODE_SMELL)

[![83.8%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '83.8%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list) [83.8% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=516&metric=new_duplicated_lines_density&view=list)



Patch:
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/DecodedRequestFrame.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/DecodedRequestFrame.java
index 865edcca..539485f3 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/DecodedRequestFrame.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/DecodedRequestFrame.java
@@ -5,9 +5,12 @@
  */
 package io.kroxylicious.proxy.frame;
 
+import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiMessage;
 
+import static org.apache.kafka.common.protocol.ApiKeys.PRODUCE;
+
 /**
  * A decoded request frame.
  */
@@ -36,4 +39,13 @@ public class DecodedRequestFrame<B extends ApiMessage>
         return decodeResponse;
     }
 
+    @Override
+    public boolean hasResponse() {
+        return !isZeroAcksProduceRequest();
+    }
+
+    private boolean isZeroAcksProduceRequest() {
+        return apiKey() == PRODUCE && ((ProduceRequestData) body).acks() == 0;
+    }
+
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/OpaqueRequestFrame.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/OpaqueRequestFrame.java
index cee64f72..ac3a114d 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/OpaqueRequestFrame.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/OpaqueRequestFrame.java
@@ -12,19 +12,23 @@ import io.netty.buffer.ByteBuf;
 public class OpaqueRequestFrame extends OpaqueFrame implements RequestFrame {
 
     private final boolean decodeResponse;
+    private boolean hasResponse;
 
     /**
      * @param buf The message buffer (excluding the frame size)
      * @param correlationId
      * @param decodeResponse
      * @param length
+     * @param hasResponse
      */
     public OpaqueRequestFrame(ByteBuf buf,
                               int correlationId,
                               boolean decodeResponse,
-                              int length) {
+                              int length,
+                              boolean hasResponse) {
         super(buf, correlationId, length);
         this.decodeResponse = decodeResponse;
+        this.hasResponse = hasResponse;
     }
 
     @Override
@@ -32,6 +36,11 @@ public class OpaqueRequestFrame extends OpaqueFrame implements RequestFrame {
         return decodeResponse;
     }
 
+    @Override
+    public boolean hasResponse() {
+        return hasResponse;
+    }
+
     @Override
     public String toString() {
         int index = buf.readerIndex();
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/RequestFrame.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/RequestFrame.java
index a80d01c0..7262968b 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/RequestFrame.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/frame/RequestFrame.java
@@ -13,4 +13,12 @@ public interface RequestFrame extends Frame {
      */
     boolean decodeResponse();
 
+    /**
+     * Whether the Kafka Client expects a response to this request
+     * @return Whether the Kafka Client expects a response to this request
+     */
+    default boolean hasResponse() {
+        return true;
+    }
+
 }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
index 061ea7ef..c419b617 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java
@@ -149,10 +149,7 @@ public class FilterHandler extends ChannelDuplexHandler {
 
             if (requestFilterResult.message() != null) {
                 if (requestFilterResult.shortCircuitResponse()) {
-                    // this is the short circuit path
-                    var header = requestFilterResult.header() == null ? new ResponseHeaderData() : ((ResponseHeaderData) requestFilterResult.header());
-                    header.setCorrelationId(decodedFrame.correlationId());
-                    filterContext.forwardResponseInternal(header, requestFilterResult.message());
+                    forwardShortCircuitResponse(ctx, decodedFrame, filterContext, requestFilterResult);
                 }
                 else {
                     var header = requestFilterResult.header() == null ? decodedFrame.header() : requestFilterResult.header();
@@ -183,6 +180,22 @@ public class FilterHandler extends ChannelDuplexHandler {
         });
     }
 
+    private void forwardShortCircuitResponse(ChannelHandlerContext ctx, DecodedRequestFrame<?> decodedFrame, DefaultFilterContext filterContext,
+                                             RequestFilterResult requestFilterResult) {
+        if (decodedFrame.hasResponse()) {
+            var header = requestFilterResult.header() == null ? new ResponseHeaderData() : ((ResponseHeaderData) requestFilterResult.header());
+            header.setCorrelationId(decodedFrame.correlationId());
+            filterContext.forwardResponseInternal(header, requestFilterResult.message());
+        }
+        else {
+            if (LOGGER.isDebugEnabled()) {
+                LOGGER.debug("{}: Filter {} attempted to short-circuit respond to a message with apiKey {}" +
+                        " that has no response in the Kafka Protocol, dropping response",
+                        ctx.channel(), filterDescriptor(), decodedFrame.apiKey());
+            }
+        }
+    }
+
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
         if (msg instanceof InternalResponseFrame<?> decodedFrame) {
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java
index 281d0c14..28a6cf81 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java
@@ -51,6 +51,7 @@ public class KafkaRequestDecoder extends KafkaMessageDecoder {
         if (log().isTraceEnabled()) { // avoid boxing
             log().trace("{}: apiVersion: {}", ctx, apiVersion);
         }
+        final int startOfMessage = in.readerIndex();
         int correlationId = in.readInt();
         LOGGER.debug("{}: {} downstream correlation id: {}", ctx, apiKey, correlationId);
 
@@ -89,29 +90,83 @@ public class KafkaRequestDecoder extends KafkaMessageDecoder {
                 log().trace("{}: body {}", ctx, body);
             }
 
-            frame = new DecodedRequestFrame<ApiMessage>(apiVersion, correlationId, decodeResponse, header, body);
+            frame = new DecodedRequestFrame<>(apiVersion, correlationId, decodeResponse, header, body);
             if (log().isTraceEnabled()) {
                 log().trace("{}: frame {}", ctx, frame);
             }
         }
         else {
+            boolean hasResponse = true;
+            if (apiKey == ApiKeys.PRODUCE) {
+                short acks = readAcks(in, startOfMessage, apiKey.id, apiVersion);
+                hasResponse = acks != 0;
+            }
             in.readerIndex(sof);
-            frame = opaqueFrame(in, correlationId, decodeResponse, length);
+            frame = opaqueFrame(in, correlationId, decodeResponse, length, hasResponse);
             in.readerIndex(sof + length);
         }
 
         return frame;
     }
 
+    private static void incrementReaderIndex(ByteBuf byteBuf, int increment) {
+        byteBuf.readerIndex(byteBuf.readerIndex() + increment);
+    }
+
+    static short readAcks(ByteBuf in, int startOfMessage, short apiKey, short apiVersion) {
+        // Annoying case: we need to know whether to expect a response so that we know
+        // whether to add to the correlation (so that, in turn, we know how to rewrite the correlation
+        // id of the client response).
+        // Adding ack-less Produce requests to the correlation => OOME.
+        // This requires decoding at least the first one or two
+        // fields of all Produce requests.
+        // Because we want to avoid parsing the produce request using ProduceRequestData
+        // just for this we are stuck with hand coding deserialization code...
+        in.readerIndex(startOfMessage);
+        short headerVersion = ApiKeys.forId(apiKey).requestHeaderVersion(apiVersion);
+        incrementReaderIndex(in, 4);
+        if (headerVersion >= 1) {
+            int clientIdLength = in.readShort();
+            incrementReaderIndex(in, clientIdLength);
+        }
+        if (headerVersion >= 2) {
+            int numTaggedFields = ByteBufAccessorImpl.readUnsignedVarint(in);
+            for (int i = 0; i < numTaggedFields; i++) {
+                ByteBufAccessorImpl.readUnsignedVarint(in);
+                int size = ByteBufAccessorImpl.readUnsignedVarint(in);
+                incrementReaderIndex(in, size);
+            }
+        }
+
+        final short acks;
+        if (apiVersion >= 3) { // Transactional id comes before acks
+            int transactionIdLength;
+            if (apiVersion < 9) { // Last non-flexible version
+                transactionIdLength = in.readShort();
+            }
+            else if (apiVersion == 9) { // First flexible version
+                transactionIdLength = ByteBufAccessorImpl.readUnsignedVarint(in);
+            }
+            else {
+                throw new AssertionError("Unsupported Produce apiVersion: " + apiVersion);
+            }
+            incrementReaderIndex(in, transactionIdLength);
+        }
+        acks = in.readShort();
+        return acks;
+    }
+
     private OpaqueRequestFrame opaqueFrame(ByteBuf in,
                                            int correlationId,
                                            boolean decodeResponse,
-                                           int length) {
+                                           int length,
+                                           boolean hasResponse) {
         return new OpaqueRequestFrame(
                 in.readSlice(length).retain(),
                 correlationId,
                 decodeResponse,
-                length);
+                length,
+                hasResponse);
     }
 
     private RequestHeaderData readHeader(short headerVersion, Readable accessor) {
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestEncoder.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestEncoder.java
index 24f3f54e..9cf7efda 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestEncoder.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestEncoder.java
@@ -5,7 +5,6 @@
  */
 package io.kroxylicious.proxy.internal.codec;
 
-import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -13,7 +12,6 @@ import org.slf4j.LoggerFactory;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 
-import io.kroxylicious.proxy.frame.DecodedRequestFrame;
 import io.kroxylicious.proxy.frame.RequestFrame;
 import io.kroxylicious.proxy.internal.InternalRequestFrame;
 
@@ -45,7 +43,7 @@ public class KafkaRequestEncoder extends KafkaMessageEncoder<RequestFrame> {
         out.readerIndex(LENGTH);
         short apiKey = out.readShort();
         short apiVersion = out.readShort();
-        boolean hasResponse = hasResponse(frame, out, ri, apiKey, apiVersion);
+        boolean hasResponse = frame.hasResponse();
         boolean decodeResponse = frame.decodeResponse();
         int downstreamCorrelationId = frame.correlationId();
         int upstreamCorrelationId = correlationManager.putBrokerRequest(apiKey,
@@ -70,68 +68,4 @@ public class KafkaRequestEncoder extends KafkaMessageEncoder<RequestFrame> {
         }
     }
 
-    private boolean hasResponse(RequestFrame frame, ByteBuf out, int ri, short apiKey, short apiVersion) {
-        if (frame instanceof DecodedRequestFrame) {
-            return apiKey != ApiKeys.PRODUCE.id
-                    || ((ProduceRequestData) ((DecodedRequestFrame<?>) frame).body()).acks() != 0;
-        }
-        else {
-            return apiKey != ApiKeys.PRODUCE.id
-                    || readAcks(out, ri, apiKey, apiVersion) != 0;
-        }
-    }
-
-    static short readAcks(ByteBuf in, int startOfMessage, short apiKey, short apiVersion) {
-        // Annoying case: we need to know whether to expect a response so that we know
-        // whether to add to the correlation (so that, in turn, we know how to rewrite the correlation
-        // id of the client response).
-        // Adding ack-less Produce requests to the correlation => OOME.
-        // This requires decoding at least the first one or two
-        // fields of all Produce requests.
-        // Because we want to avoid parsing the produce request using ProduceRequestData
-        // just for this we are stuck with hand coding deserialization code...
-        // final int keyVersionAndCorrIdSize = 8;
-        // final int startOfBodyOffset;
-        short headerVersion = ApiKeys.forId(apiKey).requestHeaderVersion(apiVersion);
-        incrementReaderIndex(in, 4);
-        if (headerVersion >= 1) {
-            int clientIdLength = in.readShort();
-            incrementReaderIndex(in, clientIdLength);
-        }
-        if (headerVersion >= 2) {
-            int numTaggedFields = ByteBufAccessorImpl.readUnsignedVarint(in);
-            for (int i = 0; i < numTaggedFields; i++) {
-                int tag = ByteBufAccessorImpl.readUnsignedVarint(in);
-                int size = ByteBufAccessorImpl.readUnsignedVarint(in);
-                incrementReaderIndex(in, size);
-            }
-        }
-
-        final short acks;
-        if (apiVersion < 3) {
-            acks = in.readShort();
-        }
-        else { // Transactional id comes before acks
-            int transactionIdLength;
-            if (apiVersion < 9) { // Last non-flexible version
-                transactionIdLength = in.readShort();
-            }
-            else if (apiVersion <= 9) { // First flexible version
-                transactionIdLength = ByteBufAccessorImpl.readUnsignedVarint(in);
-            }
-            else {
-                throw new AssertionError("Unsupported Produce apiVersion: " + apiVersion);
-            }
-            incrementReaderIndex(in, transactionIdLength);
-            acks = in.readShort();
-        }
-        // Reset index
-        in.readerIndex(startOfMessage);
-        return acks;
-    }
-
-    private static void incrementReaderIndex(ByteBuf byteBuf, int increment) {
-        byteBuf.readerIndex(byteBuf.readerIndex() + increment);
-    }
-
 }


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
index 3f0f1605..7e43dd57 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java
@@ -17,6 +17,7 @@ import java.util.concurrent.TimeUnit;
 
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.errors.InvalidTopicException;
@@ -54,6 +55,7 @@ import static io.kroxylicious.test.tester.KroxyliciousTesters.kroxyliciousTester
 import static io.kroxylicious.test.tester.KroxyliciousTesters.mockKafkaKroxyliciousTester;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.ACKS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.CLIENT_ID_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG;
 import static org.apache.kafka.common.protocol.ApiKeys.LIST_GROUPS;
@@ -305,6 +307,53 @@ class KrpcFilterIT {
         }
     }
 
+    // zero-ack produce requests require special handling because they have no response associated
+    // this checks that Kroxy can handle the basics of forwarding them.
+    @Test
+    public void shouldModifyZeroAckProduceMessage(KafkaCluster cluster, Admin admin) throws Exception {
+        admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get();
+
+        var config = proxy(cluster)
+                .addToFilters(new FilterDefinitionBuilder("ProduceRequestTransformation").withConfig("transformation", TestEncoder.class.getName()).build());
+
+        try (var tester = kroxyliciousTester(config);
+                var producer = tester.producer(Map.of(CLIENT_ID_CONFIG, "shouldModifyProduceMessage", DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000, ACKS_CONFIG, "0"));
+                var consumer = tester
+                        .consumer(Serdes.String(), Serdes.ByteArray(), Map.of(GROUP_ID_CONFIG, "my-group-id", AUTO_OFFSET_RESET_CONFIG, "earliest"))) {
+            producer.send(new ProducerRecord<>(TOPIC_1, "my-key", PLAINTEXT)).get();
+            producer.flush();
+
+            ConsumerRecords<String, byte[]> records1;
+            consumer.subscribe(Set.of(TOPIC_1));
+            records1 = consumer.poll(Duration.ofSeconds(10));
+
+            assertThat(records1).hasSize(1);
+            assertThat(records1.records(TOPIC_1)).map(ConsumerRecord::value)
+                    .containsExactly(TOPIC_1_CIPHERTEXT);
+        }
+    }
+
+    @Test
+    public void shouldForwardUnfilteredZeroAckProduceMessage(KafkaCluster cluster, Admin admin) throws Exception {
+        admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get();
+
+        var config = proxy(cluster);
+
+        try (var tester = kroxyliciousTester(config);
+                var producer = tester.producer(Map.of(CLIENT_ID_CONFIG, "shouldModifyProduceMessage", DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000, ACKS_CONFIG, "0"));
+                var consumer = tester
+                        .consumer(Serdes.String(), Serdes.String(), Map.of(GROUP_ID_CONFIG, "my-group-id", AUTO_OFFSET_RESET_CONFIG, "earliest"))) {
+            producer.send(new ProducerRecord<>(TOPIC_1, "my-key", PLAINTEXT)).get();
+            producer.flush();
+
+            consumer.subscribe(Set.of(TOPIC_1));
+            var records1 = consumer.poll(Duration.ofSeconds(10));
+
+            assertEquals(1, records1.count());
+            assertEquals(PLAINTEXT, records1.iterator().next().value());
+        }
+    }
+
     @Test
     void shouldModifyFetchMessage(KafkaCluster cluster, Admin admin) throws Exception {
 
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
index c55d7b30..bd73dd28 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/FilterHandlerTest.java
@@ -20,6 +20,7 @@ import org.apache.kafka.common.message.ApiVersionsResponseData;
 import org.apache.kafka.common.message.FetchRequestData;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceResponseData;
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiMessage;
@@ -33,6 +34,7 @@ import org.junit.jupiter.params.provider.ValueSource;
 import io.kroxylicious.proxy.filter.ApiVersionsRequestFilter;
 import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
 import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.filter.ProduceRequestFilter;
 import io.kroxylicious.proxy.filter.RequestFilterResult;
 import io.kroxylicious.proxy.filter.ResponseFilterResult;
 import io.kroxylicious.proxy.frame.DecodedRequestFrame;
@@ -65,6 +67,40 @@ public class FilterHandlerTest extends FilterHarness {
         assertEquals(frame, propagated, "Expect it to be the frame that was sent");
     }
 
+    @Test
+    void testShortCircuitResponse() {
+        ApiVersionsResponseData responseData = new ApiVersionsResponseData();
+        ApiVersionsRequestFilter filter = (apiVersion, header, request, context) -> context.requestFilterResultBuilder().shortCircuitResponse(responseData)
+                .completed();
+        buildChannel(filter);
+        writeRequest(new ApiVersionsRequestData());
+        DecodedResponseFrame<?> propagated = channel.readInbound();
+        assertEquals(responseData, propagated.body(), "expected ApiVersionsResponseData to be forwarded");
+    }
+
+    // We want to prevent a Kroxylicious filter unexpectedly responding to zero-ack produce requests
+    @Test
+    void testShortCircuitResponseZeroAcksProduceResponseDropped() {
+        ProduceResponseData responseData = new ProduceResponseData();
+        ProduceRequestFilter filter = (apiVersion, header, request, context) -> context.requestFilterResultBuilder().shortCircuitResponse(responseData)
+                .completed();
+        buildChannel(filter);
+        writeRequest(new ProduceRequestData().setAcks((short) 0));
+        DecodedResponseFrame<?> propagated = channel.readInbound();
+        assertNull(propagated, "no message should have been propagated");
+    }
+
+    @Test
+    void testShortCircuitResponseNonZeroAcksProduceResponseForwarded() {
+        ProduceResponseData responseData = new ProduceResponseData();
+        ProduceRequestFilter filter = (apiVersion, header, request, context) -> context.requestFilterResultBuilder().shortCircuitResponse(responseData)
+                .completed();
+        buildChannel(filter);
+        writeRequest(new ProduceRequestData().setAcks((short) 1));
+        DecodedResponseFrame<?> propagated = channel.readInbound();
+        assertEquals(responseData, propagated.body(), "expected ProduceResponseData to be forwarded");
+    }
+
     @Test
     void testDeferredRequestTimeout() {
         var filterFuture = new CompletableFuture<RequestFilterResult>();
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/AbstractCodecTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/AbstractCodecTest.java
index 41bd4107..7ad3c6a0 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/AbstractCodecTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/AbstractCodecTest.java
@@ -36,6 +36,7 @@ import io.netty.channel.ChannelHandlerContext;
 import io.kroxylicious.proxy.frame.DecodedFrame;
 import io.kroxylicious.proxy.frame.Frame;
 import io.kroxylicious.proxy.frame.OpaqueFrame;
+import io.kroxylicious.proxy.frame.RequestFrame;
 
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -180,7 +181,8 @@ public abstract class AbstractCodecTest {
                                                                                            BiFunction<Short, ByteBuffer, B> bodyDeser,
                                                                                            KafkaMessageDecoder decoder,
                                                                                            Class<? extends DecodedFrame> frameClass,
-                                                                                           Function<H, H> headerAdjuster) {
+                                                                                           Function<H, H> headerAdjuster,
+                                                                                           boolean expectedHasResponse) {
 
         ApiMessage encodedHeader = headerSupplier.apply(apiVersion);
 
@@ -209,6 +211,8 @@ public abstract class AbstractCodecTest {
         DecodedFrame<H, B> frame = (DecodedFrame<H, B>) messages.get(0);
         assertEquals(akHeader, headerAdjuster.apply(frame.header()));
         assertEquals(akBody, frame.body());
+        boolean hasResponse = frame instanceof RequestFrame requestFrame && requestFrame.hasResponse();
+        assertEquals(expectedHasResponse, hasResponse);
         return frame.correlationId();
     }
 
@@ -218,7 +222,8 @@ public abstract class AbstractCodecTest {
                                                                                            Function<Short, H> headerSupplier,
                                                                                            Supplier<B> bodySupplier,
                                                                                            KafkaMessageDecoder decoder,
-                                                                                           Class<? extends Frame> frameClass)
+                                                                                           Class<? extends Frame> frameClass,
+                                                                                           boolean expectedHasResponse)
             throws Exception {
 
         var encodedHeader = headerSupplier.apply(apiVersion);
@@ -237,6 +242,8 @@ public abstract class AbstractCodecTest {
         OpaqueFrame frame = (OpaqueFrame) messages.get(0);
         akBuf.readerIndex(4); // An OpaqueFrame excludes the length of the frame
         assertSameBytes(akBuf, frame.buf());
+        boolean hasResponse = frame instanceof RequestFrame requestFrame && requestFrame.hasResponse();
+        assertEquals(expectedHasResponse, hasResponse);
         return frame.correlationId();
     }
 
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java
index 78525da3..f054dfc9 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestDecoderTest.java
@@ -9,10 +9,15 @@ import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletionStage;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
+import org.apache.kafka.common.message.ApiMessageType;
 import org.apache.kafka.common.message.ApiVersionsRequestData;
+import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 
@@ -30,9 +35,44 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class RequestDecoderTest extends AbstractCodecTest {
 
+    public static final DecodePredicate DECODE_NOTHING = new DecodePredicate() {
+        @Override
+        public boolean shouldDecodeRequest(ApiKeys apiKey, short apiVersion) {
+            return false;
+        }
+
+        @Override
+        public boolean shouldDecodeResponse(ApiKeys apiKey, short apiVersion) {
+            return false;
+        }
+    };
+    public static final DecodePredicate DECODE_EVERYTHING = new DecodePredicate() {
+        @Override
+        public boolean shouldDecodeRequest(ApiKeys apiKey, short apiVersion) {
+            return true;
+        }
+
+        @Override
+        public boolean shouldDecodeResponse(ApiKeys apiKey, short apiVersion) {
+            return true;
+        }
+    };
+
+    public static List<Object[]> produceRequestApiVersions() {
+        List<Short> produceVersions = requestApiVersions(ApiMessageType.PRODUCE).collect(Collectors.toList());
+        List<Object[]> cartesianProduct = new ArrayList<>();
+        produceVersions
+                .forEach(produceVersion -> Stream.of((short) 0, (short) 1, (short) -1).forEach(acks -> cartesianProduct.add(new Object[]{ produceVersion, acks })));
+        return cartesianProduct;
+    }
+
+    static ProduceRequestData deserializeProduceRequestUsingKafkaApis(short apiVersion, ByteBuffer buffer) {
+        return new ProduceRequestData(new ByteBufferAccessor(buffer), apiVersion);
+    }
+
     @ParameterizedTest
     @MethodSource("requestApiVersions")
-    void testApiVersionsExactlyOneFrame_decoded(short apiVersion) throws Exception {
+    void testApiVersionsExactlyOneFrame_decoded(short apiVersion) {
         assertEquals(12,
                 exactlyOneFrame_decoded(apiVersion,
                         ApiKeys.API_VERSIONS::requestHeaderVersion,
@@ -46,7 +86,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
                                                 (ApiVersionsRequestFilter) (version, header, request, context) -> context.requestFilterResultBuilder()
                                                         .forward(header, request).completed()))),
                         DecodedRequestFrame.class,
-                        (RequestHeaderData header) -> header),
+                        (RequestHeaderData header) -> header, true),
                 "Unexpected correlation id");
     }
 
@@ -73,7 +113,7 @@ public class RequestDecoderTest extends AbstractCodecTest {
                                         return context.requestFilterResultBuilder().forward(header, request).completed();
                                     }
                                 }))),
-                        OpaqueRequestFrame.class),
+                        OpaqueRequestFrame.class, true),
                 "Unexpected correlation id");
     }
 
@@ -172,4 +212,67 @@ public class RequestDecoderTest extends AbstractCodecTest {
 
         assertEquals(byteBuf.writerIndex(), byteBuf.readerIndex());
     }
+
+    /**
+     * KafkaRequestDecoder has some special case handling for Produce requests with acks==0
+     */
+    @ParameterizedTest
+    @MethodSource("produceRequestApiVersions")
+    void testAcksParsingWhenNotDecoding(short produceVersion, short acks) throws Exception {
+
+        var header = new RequestHeaderData()
+                .setRequestApiKey(ApiKeys.PRODUCE.id)
+                .setRequestApiVersion(produceVersion)
+                .setCorrelationId(45);
+        short headerVersion = ApiKeys.PRODUCE.requestHeaderVersion(produceVersion);
+        if (headerVersion >= 1) {
+            header.setClientId("323423");
+        }
+        var body = new ProduceRequestData()
+                .setAcks(acks);
+        if (produceVersion >= 3) {
+            body.setTransactionalId("wnedkwjn");
+        }
+        assertEquals(45,
+                exactlyOneFrame_encoded(produceVersion,
+                        ApiKeys.PRODUCE::requestHeaderVersion,
+                        (x) -> header,
+                        () -> body,
+                        new KafkaRequestDecoder(DECODE_NOTHING),
+                        OpaqueRequestFrame.class,
+                        acks != 0),
+                "Unexpected correlation id");
+    }
+
+    /**
+     * KafkaRequestDecoder has some special case handling for Produce requests with acks==0
+     */
+    @ParameterizedTest
+    @MethodSource("produceRequestApiVersions")
+    void testAcksParsingWhenDecoding(short produceVersion, short acks) throws Exception {
+
+        var header = new RequestHeaderData()
+                .setRequestApiKey(ApiKeys.PRODUCE.id)
+                .setRequestApiVersion(produceVersion)
+                .setCorrelationId(45);
+        short headerVersion = ApiKeys.PRODUCE.requestHeaderVersion(produceVersion);
+        if (headerVersion >= 1) {
+            header.setClientId("323423");
+        }
+        var body = new ProduceRequestData()
+                .setAcks(acks);
+        if (produceVersion >= 3) {
+            body.setTransactionalId("wnedkwjn");
+        }
+        assertEquals(45,
+                exactlyOneFrame_decoded(produceVersion,
+                        ApiKeys.PRODUCE::requestHeaderVersion,
+                        (x) -> header,
+                        () -> body,
+                        AbstractCodecTest::deserializeRequestHeaderUsingKafkaApis,
+                        RequestDecoderTest::deserializeProduceRequestUsingKafkaApis,
+                        new KafkaRequestDecoder(DECODE_EVERYTHING),
+                        DecodedRequestFrame.class, ((RequestHeaderData head) -> head), acks != 0),
+                "Unexpected correlation id");
+    }
 }
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestEncoderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestEncoderTest.java
index b24c5c59..2d60f493 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestEncoderTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/RequestEncoderTest.java
@@ -8,14 +8,13 @@ package io.kroxylicious.proxy.internal.codec;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import org.apache.kafka.common.message.ApiMessageType;
 import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
+import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 
@@ -31,6 +30,16 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class RequestEncoderTest extends AbstractCodecTest {
 
+    public static List<Object[]> testResponseFrames() {
+        List<Object[]> cartesianProduct = new ArrayList<>();
+        List.of(true, false).forEach(
+                requestHasResponse -> Stream.of(true, false)
+                        .forEach(decodeResponse -> cartesianProduct.add(new Object[]{ requestHasResponse, decodeResponse })));
+        return cartesianProduct;
+    }
+
+    // TODO test API_VERSIONS header is v0
+
     @ParameterizedTest
     @MethodSource("requestApiVersions")
     void testApiVersions(short apiVersion) throws Exception {
@@ -50,23 +59,47 @@ public class RequestEncoderTest extends AbstractCodecTest {
         assertEquals(exampleHeader.correlationId(), corr.downstreamCorrelationId());
     }
 
-    // TODO test API_VERSIONS header is v0
+    /**
+     * Zero-ack produce requests do not have a response, so we do not want to add correlations in memory
+     * that will never be used. We determine this in the KafkaRequestDecoder and set a flag on the frame
+     * when there is no response expected.
+     */
+    @ParameterizedTest
+    @MethodSource("testResponseFrames")
+    void testRequestsWithNoResponseArentStoredInCorrelationManager() throws Exception {
 
-    public static List<Object[]> produceRequestApiVersions() {
-        List<Short> produceVersions = requestApiVersions(ApiMessageType.PRODUCE).collect(Collectors.toList());
-        List<Object[]> cartesianProduct = new ArrayList<>();
-        produceVersions.forEach(produceVersion -> Stream.of((short) 0, (short) 1, (short) -1)
-                .forEach(acks -> Stream.of(true, false).forEach(decodeResponse -> cartesianProduct.add(new Object[]{ produceVersion, acks, decodeResponse }))));
-        return cartesianProduct;
+        givenRequestFrame frame = createRequestFrame(false);
+
+        var correlationManager = new CorrelationManager(78);
+        whenRequestEncoded(frame, correlationManager);
+
+        assertTrue(correlationManager.brokerRequests.isEmpty(),
+                "Expect request with no response to not have a correlation stored");
     }
 
-    /**
-     * KafkaRequestEncoder has some special case handling for Produce requests with acks==0
-     */
     @ParameterizedTest
-    @MethodSource("produceRequestApiVersions")
-    void testAcksParsing(short produceVersion, short acks, boolean decodeResponse) throws Exception {
+    @MethodSource("testResponseFrames")
+    void testRequestsWithResponseAreStoredInCorrelationManager() throws Exception {
+
+        givenRequestFrame frame = createRequestFrame(true);
+
+        var correlationManager = new CorrelationManager(78);
+        whenRequestEncoded(frame, correlationManager);
+
+        assertNotNull(correlationManager.brokerRequests.get(78),
+                "Expect request with response to have a correlation stored");
+        assertEquals(1, correlationManager.brokerRequests.size(),
+                "Expect request with response to have a correlation");
+    }
+
+    private static void whenRequestEncoded(givenRequestFrame result, CorrelationManager correlationManager) throws Exception {
+        ByteBuf out = Unpooled.buffer(result.byteBuffer().capacity() + 4);
+        new KafkaRequestEncoder(correlationManager).encode(null, result.frame(), out);
+    }
 
+    @NotNull
+    private static givenRequestFrame createRequestFrame(boolean hasResponse) {
+        short produceVersion = ApiKeys.PRODUCE.latestVersion();
         var header = new RequestHeaderData()
                 .setRequestApiKey(ApiKeys.PRODUCE.id)
                 .setRequestApiVersion(produceVersion)
@@ -76,7 +109,7 @@ public class RequestEncoderTest extends AbstractCodecTest {
             header.setClientId("323423");
         }
         var body = new ProduceRequestData()
-                .setAcks(acks);
+                .setAcks((short) 0);
         if (produceVersion >= 3) {
             body.setTransactionalId("wnedkwjn");
         }
@@ -86,23 +119,11 @@ public class RequestEncoderTest extends AbstractCodecTest {
 
         ByteBuf buf = Unpooled.copiedBuffer(byteBuffer);
 
-        var frame = new OpaqueRequestFrame(buf, 12, decodeResponse, frameSize);
-
-        ByteBuf out = Unpooled.buffer(byteBuffer.capacity() + 4);
-
-        var correlationManager = new CorrelationManager(78);
-
-        new KafkaRequestEncoder(correlationManager).encode(null, frame, out);
+        var frame = new OpaqueRequestFrame(buf, 12, true, frameSize, hasResponse);
+        givenRequestFrame result = new givenRequestFrame(byteBuffer, frame);
+        return result;
+    }
 
-        if (acks == 0) {
-            assertTrue(correlationManager.brokerRequests.isEmpty(),
-                    "Expect acks == 0 to not have a correlation");
-        }
-        else {
-            assertNotNull(correlationManager.brokerRequests.get(78),
-                    "Expect acks != 0 to have a correlation");
-            assertEquals(1, correlationManager.brokerRequests.size(),
-                    "Expect acks != 0 to have a correlation");
-        }
+    private record givenRequestFrame(ByteBuffer byteBuffer, OpaqueRequestFrame frame) {
     }
 }
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/ResponseDecoderTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/ResponseDecoderTest.java
index 4cca71ca..7e9aa53c 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/ResponseDecoderTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/codec/ResponseDecoderTest.java
@@ -29,7 +29,7 @@ class ResponseDecoderTest extends AbstractCodecTest {
                 AbstractCodecTest::deserializeApiVersionsResponseUsingKafkaApis,
                 new KafkaResponseDecoder(mgr),
                 DecodedResponseFrame.class,
-                header -> header.setCorrelationId(12)),
+                header -> header.setCorrelationId(12), false),
                 "Unexpected correlation id");
     }
 
@@ -43,7 +43,7 @@ class ResponseDecoderTest extends AbstractCodecTest {
                 v -> AbstractCodecTest.exampleResponseHeader(),
                 AbstractCodecTest::exampleApiVersionsResponse,
                 new KafkaResponseDecoder(mgr),
-                OpaqueResponseFrame.class),
+                OpaqueResponseFrame.class, false),
                 "Unexpected correlation id");
     }