KafkaSinkCluster: error on AllocateProducerIds (#1804)

Co-authored-by: Conor <conor.brosnan@instaclustr.com>

.github/workflows/build_and_test.yaml

@@ -43,9 +43,6 @@ jobs:
         # Otherwise only the last build to finish would get saved to the cache.
         # We allow different test_flags to share a cache as they should have identical build outputs
         key: ${{ matrix.runner }} - ${{ matrix.cargo_flags }}
-        cache-directories: |
-          target/debug/jassets
-          target/release/jassets
 
         # this line means that only the main branch writes to the cache
         # benefits:
@@ -65,6 +62,18 @@ jobs:
       uses: taiki-e/install-action@v2
       with:
         tool: nextest@0.9.70
+      # It is currently impossible to combine j4rs, rust-cache and nextest:
+      # * j4rs needs to store jars somewhere, by default this is in target/debug/jassets, we do have the option to move this somewhere outside of target.
+      # * rust-cache will delete all files in target/debug other than `build`, `deps` and `.fingerprint`
+      # * nextest will only archive files within the target directory
+      # There is no way of combining all of these requirements.
+      # We will need to find one of these projects that is suitable to have its requirements loosened.
+      # This is going to be tricky and require discussion with the various upstream projects, and will take a while to land anything.
+      # So for now we need a quick workaround.
+      # This workaround is to force j4rs to be rebuilt from scratch via cargo clean.
+      # This has a cost on CI runtime and in the future we should find another solution as discussed above.
+    - name: Workaround j4rs cache issue
+      run: cargo clean -p j4rs
     - name: Build tests
       run: |
         cargo test --doc ${{ matrix.cargo_flags }} --all-features -- --show-output --nocapture

shotover/src/transforms/kafka/sink_cluster/mod.rs

@@ -1002,17 +1002,20 @@ impl KafkaSinkCluster {
                 }
                 Some(Frame::Kafka(KafkaFrame::Request {
                     body:
+                        // It was determined that these message types are only sent between brokers by:
+                        // * This paragraph https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=225153708#KIP866ZooKeepertoKRaftMigration-ControllerRPCs
+                        // * This field containing only "zkBroker" https://github.com/apache/kafka/blob/e3f953483cb480631bf041698770b47ddb82796f/clients/src/main/resources/common/message/LeaderAndIsrRequest.json#L19
                         RequestBody::LeaderAndIsr(_)
                         | RequestBody::StopReplica(_)
-                        | RequestBody::UpdateMetadata(_),
+                        | RequestBody::UpdateMetadata(_)
+                        // It was determined that this message type is only sent between brokers by:
+                        // * https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=177049344#KIP730:ProducerIDgenerationinKRaftmode-PublicInterfaces
+                        | RequestBody::AllocateProducerIds(_),
                     header:
                         RequestHeader {
                             request_api_key, ..
                         },
                 })) => {
-                    // It was determined that these message types are only sent between brokers by:
-                    // * This paragraph https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=225153708#KIP866ZooKeepertoKRaftMigration-ControllerRPCs
-                    // * This field containing only "zkBroker" https://github.com/apache/kafka/blob/e3f953483cb480631bf041698770b47ddb82796f/clients/src/main/resources/common/message/LeaderAndIsrRequest.json#L19
                     return Err(anyhow!(
                         r#"Client sent request of type {request_api_key}.
 This request is used only for communication between brokers and shotover does not support proxying between brokers.