Add java kafka driver backend to kafka integration tests (#1509)

shotover · Mar 4, 2024 · 4a7003f · 4a7003f
1 parent c346e10
commit 4a7003f
Show file tree

Hide file tree

Showing 12 changed files with 692 additions and 311 deletions.
diff --git a/.config/nextest.toml b/.config/nextest.toml
@@ -1,6 +1,7 @@
 [profile.default]
 fail-fast = false
 slow-timeout = { period = '5m', terminate-after = 2 }
+archive-include = ["debug/jassets", "release/jassets"]
 
 # Overwrites profile.default when the filter matches
 [[profile.default.overrides]]

diff --git a/.github/workflows/build_and_test.yaml b/.github/workflows/build_and_test.yaml
@@ -57,10 +57,11 @@ jobs:
         key: ubuntu-20.04-packages
     - name: Install ubuntu packages
       run: shotover-proxy/build/install_ubuntu_packages.sh
-    - name: Install nextest
-      uses: taiki-e/install-action@v2
-      with:
-        tool: [email protected]
+    #- name: Install nextest
+    #  uses: taiki-e/install-action@v2
+    #  with:
+    #    tool: [email protected]
+    - run: cargo install --git https://github.com/rukai/nextest --branch archive-include-legacy cargo-nextest
     - name: Build tests
       run: |
         cargo test --doc ${{ matrix.cargo_flags }} --all-features -- --show-output --nocapture

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/shotover-proxy/benches/windsock/kafka/bench.rs b/shotover-proxy/benches/windsock/kafka/bench.rs
@@ -18,14 +18,14 @@ use shotover::transforms::kafka::sink_single::KafkaSinkSingleConfig;
 use shotover::transforms::TransformConfig;
 use std::sync::Arc;
 use std::{collections::HashMap, time::Duration};
-use test_helpers::connection::kafka::rdkafka::admin::{
+use test_helpers::connection::kafka::cpp::rdkafka::admin::{
     AdminClient, AdminOptions, NewTopic, TopicReplication,
 };
-use test_helpers::connection::kafka::rdkafka::client::DefaultClientContext;
-use test_helpers::connection::kafka::rdkafka::config::ClientConfig;
-use test_helpers::connection::kafka::rdkafka::consumer::{Consumer, StreamConsumer};
-use test_helpers::connection::kafka::rdkafka::producer::{FutureProducer, FutureRecord};
-use test_helpers::connection::kafka::rdkafka::util::Timeout;
+use test_helpers::connection::kafka::cpp::rdkafka::client::DefaultClientContext;
+use test_helpers::connection::kafka::cpp::rdkafka::config::ClientConfig;
+use test_helpers::connection::kafka::cpp::rdkafka::consumer::{Consumer, StreamConsumer};
+use test_helpers::connection::kafka::cpp::rdkafka::producer::{FutureProducer, FutureRecord};
+use test_helpers::connection::kafka::cpp::rdkafka::util::Timeout;
 use test_helpers::docker_compose::docker_compose;
 use tokio::{sync::mpsc::UnboundedSender, task::JoinHandle, time::Instant};
 use windsock::{Bench, BenchParameters, Profiling, Report};

diff --git a/shotover-proxy/tests/kafka_int_tests/mod.rs b/shotover-proxy/tests/kafka_int_tests/mod.rs
@@ -1,26 +1,23 @@
-#[cfg(feature = "rdkafka-driver-tests")]
+mod test_cases;
+
 use crate::shotover_process;
-#[cfg(feature = "rdkafka-driver-tests")]
+use rstest::rstest;
 use std::time::Duration;
-#[cfg(feature = "rdkafka-driver-tests")]
+use test_helpers::connection::kafka::{KafkaConnectionBuilder, KafkaDriver};
 use test_helpers::docker_compose::docker_compose;
 
-#[cfg(feature = "rdkafka-driver-tests")]
-mod test_cases;
-
-#[cfg(feature = "rdkafka-driver-tests")]
-use test_helpers::connection::kafka::KafkaConnectionBuilder;
-
-#[cfg(feature = "rdkafka-driver-tests")]
+#[rstest]
+#[cfg_attr(feature = "rdkafka-driver-tests", case::cpp(KafkaDriver::Cpp))]
+#[case::java(KafkaDriver::Java)]
 #[tokio::test]
-async fn passthrough_standard() {
+async fn passthrough_standard(#[case] driver: KafkaDriver) {
     let _docker_compose =
         docker_compose("tests/test-configs/kafka/passthrough/docker-compose.yaml");
     let shotover = shotover_process("tests/test-configs/kafka/passthrough/topology.yaml")
         .start()
         .await;
 
-    let connection_builder = KafkaConnectionBuilder::new("127.0.0.1:9192");
+    let connection_builder = KafkaConnectionBuilder::new(driver, "127.0.0.1:9192");
     test_cases::basic(connection_builder).await;
 
     tokio::time::timeout(
@@ -31,9 +28,11 @@ async fn passthrough_standard() {
     .expect("Shotover did not shutdown within 10s");
 }
 
-#[cfg(feature = "rdkafka-driver-tests")]
+#[rstest]
+#[cfg_attr(feature = "rdkafka-driver-tests", case::cpp(KafkaDriver::Cpp))]
+#[case::java(KafkaDriver::Java)]
 #[tokio::test]
-async fn passthrough_tls() {
+async fn passthrough_tls(#[case] driver: KafkaDriver) {
     test_helpers::cert::generate_kafka_test_certs();
 
     let _docker_compose =
@@ -42,7 +41,7 @@ async fn passthrough_tls() {
         .start()
         .await;
 
-    let connection_builder = KafkaConnectionBuilder::new("127.0.0.1:9192");
+    let connection_builder = KafkaConnectionBuilder::new(driver, "127.0.0.1:9192");
     test_cases::basic(connection_builder).await;
 
     tokio::time::timeout(
@@ -53,9 +52,11 @@ async fn passthrough_tls() {
     .expect("Shotover did not shutdown within 10s");
 }
 
-#[cfg(feature = "rdkafka-driver-tests")]
+#[rstest]
+#[cfg_attr(feature = "rdkafka-driver-tests", case::cpp(KafkaDriver::Cpp))]
+#[case::java(KafkaDriver::Java)]
 #[tokio::test]
-async fn cluster_tls() {
+async fn cluster_tls(#[case] driver: KafkaDriver) {
     test_helpers::cert::generate_kafka_test_certs();
 
     let _docker_compose =
@@ -64,7 +65,7 @@ async fn cluster_tls() {
         .start()
         .await;
 
-    let connection_builder = KafkaConnectionBuilder::new("127.0.0.1:9192");
+    let connection_builder = KafkaConnectionBuilder::new(driver, "127.0.0.1:9192");
     test_cases::basic(connection_builder).await;
 
     tokio::time::timeout(
@@ -75,40 +76,46 @@ async fn cluster_tls() {
     .expect("Shotover did not shutdown within 10s");
 }
 
-#[cfg(feature = "rdkafka-driver-tests")]
+#[rstest]
+#[cfg_attr(feature = "rdkafka-driver-tests", case::cpp(KafkaDriver::Cpp))]
+#[case::java(KafkaDriver::Java)]
 #[tokio::test]
-async fn passthrough_encode() {
+async fn passthrough_encode(#[case] driver: KafkaDriver) {
     let _docker_compose =
         docker_compose("tests/test-configs/kafka/passthrough/docker-compose.yaml");
     let shotover = shotover_process("tests/test-configs/kafka/passthrough/topology-encode.yaml")
         .start()
         .await;
 
-    let connection_builder = KafkaConnectionBuilder::new("127.0.0.1:9192");
+    let connection_builder = KafkaConnectionBuilder::new(driver, "127.0.0.1:9192");
     test_cases::basic(connection_builder).await;
 
     shotover.shutdown_and_then_consume_events(&[]).await;
 }
 
-#[cfg(feature = "rdkafka-driver-tests")]
+#[rstest]
+#[cfg_attr(feature = "rdkafka-driver-tests", case::cpp(KafkaDriver::Cpp))]
+#[case::java(KafkaDriver::Java)]
 #[tokio::test]
-async fn passthrough_sasl() {
+async fn passthrough_sasl(#[case] driver: KafkaDriver) {
     let _docker_compose =
         docker_compose("tests/test-configs/kafka/passthrough-sasl/docker-compose.yaml");
     let shotover = shotover_process("tests/test-configs/kafka/passthrough-sasl/topology.yaml")
         .start()
         .await;
 
     let connection_builder =
-        KafkaConnectionBuilder::new("127.0.0.1:9192").use_sasl("user", "password");
+        KafkaConnectionBuilder::new(driver, "127.0.0.1:9192").use_sasl("user", "password");
     test_cases::basic(connection_builder).await;
 
     shotover.shutdown_and_then_consume_events(&[]).await;
 }
 
-#[cfg(feature = "rdkafka-driver-tests")]
+#[rstest]
+#[cfg_attr(feature = "rdkafka-driver-tests", case::cpp(KafkaDriver::Cpp))]
+#[case::java(KafkaDriver::Java)]
 #[tokio::test]
-async fn passthrough_sasl_encode() {
+async fn passthrough_sasl_encode(#[case] driver: KafkaDriver) {
     let _docker_compose =
         docker_compose("tests/test-configs/kafka/passthrough-sasl/docker-compose.yaml");
     let shotover =
@@ -117,22 +124,24 @@ async fn passthrough_sasl_encode() {
             .await;
 
     let connection_builder =
-        KafkaConnectionBuilder::new("127.0.0.1:9192").use_sasl("user", "password");
+        KafkaConnectionBuilder::new(driver, "127.0.0.1:9192").use_sasl("user", "password");
     test_cases::basic(connection_builder).await;
 
     shotover.shutdown_and_then_consume_events(&[]).await;
 }
 
-#[cfg(feature = "rdkafka-driver-tests")]
+#[rstest]
+#[cfg_attr(feature = "rdkafka-driver-tests", case::cpp(KafkaDriver::Cpp))]
+#[case::java(KafkaDriver::Java)]
 #[tokio::test]
-async fn cluster_1_rack_single_shotover() {
+async fn cluster_1_rack_single_shotover(#[case] driver: KafkaDriver) {
     let _docker_compose =
         docker_compose("tests/test-configs/kafka/cluster-1-rack/docker-compose.yaml");
     let shotover = shotover_process("tests/test-configs/kafka/cluster-1-rack/topology-single.yaml")
         .start()
         .await;
 
-    let connection_builder = KafkaConnectionBuilder::new("127.0.0.1:9192");
+    let connection_builder = KafkaConnectionBuilder::new(driver, "127.0.0.1:9192");
     test_cases::basic(connection_builder).await;
 
     tokio::time::timeout(
@@ -143,9 +152,11 @@ async fn cluster_1_rack_single_shotover() {
     .expect("Shotover did not shutdown within 10s");
 }
 
-#[cfg(feature = "rdkafka-driver-tests")]
-#[tokio::test]
-async fn cluster_1_rack_multi_shotover() {
+#[rstest]
+#[cfg_attr(feature = "rdkafka-driver-tests", case::cpp(KafkaDriver::Cpp))]
+#[case::java(KafkaDriver::Java)]
+#[tokio::test(flavor = "multi_thread")]
+async fn cluster_1_rack_multi_shotover(#[case] driver: KafkaDriver) {
     let _docker_compose =
         docker_compose("tests/test-configs/kafka/cluster-1-rack/docker-compose.yaml");
     let mut shotovers = vec![];
@@ -163,7 +174,7 @@ async fn cluster_1_rack_multi_shotover() {
         );
     }
 
-    let connection_builder = KafkaConnectionBuilder::new("127.0.0.1:9192");
+    let connection_builder = KafkaConnectionBuilder::new(driver, "127.0.0.1:9192");
     test_cases::basic(connection_builder).await;
 
     for shotover in shotovers {
@@ -176,17 +187,19 @@ async fn cluster_1_rack_multi_shotover() {
     }
 }
 
-#[cfg(feature = "rdkafka-driver-tests")]
+#[rstest]
+#[cfg_attr(feature = "rdkafka-driver-tests", case::cpp(KafkaDriver::Cpp))]
+#[case::java(KafkaDriver::Java)]
 #[tokio::test]
-async fn cluster_2_racks_single_shotover() {
+async fn cluster_2_racks_single_shotover(#[case] driver: KafkaDriver) {
     let _docker_compose =
         docker_compose("tests/test-configs/kafka/cluster-2-racks/docker-compose.yaml");
     let shotover =
         shotover_process("tests/test-configs/kafka/cluster-2-racks/topology-single.yaml")
             .start()
             .await;
 
-    let connection_builder = KafkaConnectionBuilder::new("127.0.0.1:9192");
+    let connection_builder = KafkaConnectionBuilder::new(driver, "127.0.0.1:9192");
     test_cases::basic(connection_builder).await;
 
     tokio::time::timeout(
@@ -197,9 +210,11 @@ async fn cluster_2_racks_single_shotover() {
     .expect("Shotover did not shutdown within 10s");
 }
 
-#[cfg(feature = "rdkafka-driver-tests")]
+#[rstest]
+#[cfg_attr(feature = "rdkafka-driver-tests", case::cpp(KafkaDriver::Cpp))]
+#[case::java(KafkaDriver::Java)]
 #[tokio::test]
-async fn cluster_2_racks_multi_shotover() {
+async fn cluster_2_racks_multi_shotover(#[case] driver: KafkaDriver) {
     let _docker_compose =
         docker_compose("tests/test-configs/kafka/cluster-2-racks/docker-compose.yaml");
 
@@ -219,7 +234,7 @@ async fn cluster_2_racks_multi_shotover() {
         );
     }
 
-    let connection_builder = KafkaConnectionBuilder::new("127.0.0.1:9192");
+    let connection_builder = KafkaConnectionBuilder::new(driver, "127.0.0.1:9192");
     test_cases::basic(connection_builder).await;
 
     for shotover in shotovers {