Merge branch 'main' into precommit_file

.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,22 @@ jobs:
       uses: taiki-e/install-action@v2
       with:
         tool: [email protected]
+
+      # 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.
+      # I suspect that rust-cache is the project that needs to change, maybe add a config field to include extra paths in the cache, similar to nextest's archive.include.
+      # 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-proxy/tests/kafka_int_tests/test_cases.rs

@@ -133,6 +133,28 @@ async fn admin_setup(connection_builder: &KafkaConnectionBuilder) {
 async fn admin_cleanup(connection_builder: &KafkaConnectionBuilder) {
     let admin = connection_builder.connect_admin().await;
 
+    // Only supported by java driver
+    #[allow(irrefutable_let_patterns)]
+    if let KafkaConnectionBuilder::Java(_) = connection_builder {
+        // It is not clear how to actually invoke this API in a succesful way.
+        // At the very least this test case shows that shotover succesfully sends and receives this message type (even if the broker responds with an error)
+        match admin
+            .elect_leaders(&[TopicPartition {
+                topic_name: "partitions1_with_offset".to_owned(),
+                partition: 0,
+            }])
+            .await
+        {
+            Ok(()) => panic!("elect_leaders is expected to fail since an election is not required"),
+            Err(e) => {
+                assert_eq!(
+                    format!("{e}"),
+                    "org.apache.kafka.common.errors.ElectionNotNeededException: Leader election not needed for topic partition.\n"
+                );
+            }
+        }
+    }
+
     admin.delete_groups(&["some_group", "some_group1"]).await;
     delete_records_partitions1(&admin, connection_builder).await;
     delete_records_partitions3(&admin, connection_builder).await;

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

@@ -1048,6 +1048,10 @@ The connection to the client has been closed."
                     body: RequestBody::CreateTopics(_),
                     ..
                 })) => self.route_to_controller(request),
+                Some(Frame::Kafka(KafkaFrame::Request {
+                    body: RequestBody::ElectLeaders(_),
+                    ..
+                })) => self.route_to_controller(request),
 
                 // route to all nodes
                 Some(Frame::Kafka(KafkaFrame::Request {

test-helpers/src/connection/kafka/java.rs

@@ -779,6 +779,26 @@ impl KafkaAdminJava {
             .await;
     }
 
+    pub async fn elect_leaders(&self, topic_partitions: &[TopicPartition]) -> Result<()> {
+        let election_type = self
+            .jvm
+            .class("org.apache.kafka.common.ElectionType")
+            .field("PREFERRED");
+        let topic_partitions_java = self.jvm.new_set(
+            "org.apache.kafka.common.TopicPartition",
+            topic_partitions
+                .iter()
+                .map(|topic_partition| topic_partition_to_java(&self.jvm, topic_partition))
+                .collect(),
+        );
+
+        self.admin
+            .call("electLeaders", vec![election_type, topic_partitions_java])
+            .call_async_fallible("all", vec![])
+            .await
+            .map(|_| ())
+    }
+
     pub async fn create_acls(&self, acls: Vec<Acl>) {
         let resource_type = self
             .jvm

test-helpers/src/connection/kafka/mod.rs

@@ -507,6 +507,16 @@ impl KafkaAdmin {
         }
     }
 
+    pub async fn elect_leaders(&self, topic_partitions: &[TopicPartition]) -> Result<()> {
+        match self {
+            #[cfg(feature = "kafka-cpp-driver-tests")]
+            Self::Cpp(_) => {
+                panic!("rdkafka-rs driver does not support elect_leaders")
+            }
+            Self::Java(java) => java.elect_leaders(topic_partitions).await,
+        }
+    }
+
     pub async fn create_partitions(&self, partitions: &[NewPartition<'_>]) {
         match self {
             #[cfg(feature = "kafka-cpp-driver-tests")]