fix: ensure Java interopt

The only issue found by this is that we were not able to read
Snappy-compressed data produced by Java clients, because Java uses its
own weird framing format. Funnily Java clients can read the Snappy data
what we produce. This is also in line with the rdkafka implemention
(they can consume the "normal" and the special "Java" format but only
produce the "normal" one).

Author: crepererum

.circleci/config.yml

@@ -31,14 +31,14 @@ commands:
             - /usr/local/cargo/registry
           key: cargo-cache-{{ arch }}-{{ .Branch }}-{{ checksum "Cargo.toml" }}
 
-  install_zlib_dev:
-    description: Install zlib-dev
+  install_packages:
+    description: Install openjdk-11-jre and zlib-dev
     steps:
       - run:
-          name: Install zlib-dev
+          name: Install openjdk-11-jre and zlib-dev
           command: |
             sudo apt-get update
-            sudo apt-get install -y zlib1g-dev
+            sudo apt-get install -y openjdk-11-jre zlib1g-dev
 
 jobs:
   fmt:
@@ -190,14 +190,15 @@ jobs:
       RUST_LOG: "trace"
       # Run integration tests
       TEST_INTEGRATION: 1
+      TEST_JAVA_INTEROPT: 1
       # Don't use the first node here since this is likely the controller and we want to ensure that we automatically
       # pick the controller for certain actions (e.g. topic creation) and don't just get lucky.
       KAFKA_CONNECT: "redpanda-1:9092"
     steps:
       - checkout
       - rust_components
       - cache_restore
-      - install_zlib_dev
+      - install_packages
       - run:
           name: Cargo test
           command: cargo test --all-features --all-targets
@@ -258,14 +259,15 @@ jobs:
       TEST_INTEGRATION: 1
       # Kafka support DeleteRecords
       TEST_DELETE_RECORDS: 1
+      TEST_JAVA_INTEROPT: 1
       # Don't use the first node here since this is likely the controller and we want to ensure that we automatically
       # pick the controller for certain actions (e.g. topic creation) and don't just get lucky.
       KAFKA_CONNECT: "kafka-1:9093"
     steps:
       - checkout
       - rust_components
       - cache_restore
-      - install_zlib_dev
+      - install_packages
       - run:
           name: Cargo test
           command: cargo test --all-features --all-targets

Cargo.toml

@@ -43,6 +43,9 @@ assert_matches = "1.5"
 criterion = { version = "0.3", features = ["async_tokio"] }
 dotenv = "0.15"
 futures = "0.3"
+j4rs = "0.13"
+once_cell = "1.9"
+procspawn = "0.10"
 proptest = "1"
 proptest-derive = "0.3"
 rustls-pemfile = "0.3"

README.md

@@ -136,6 +136,10 @@ $ TEST_INTEGRATION=1 TEST_DELETE_RECORDS=1 KAFKA_CONNECT=localhost:9094 cargo te
 in another session. Note that Apache Kafka supports a different set of features then redpanda, so we pass other
 environment variables.
 
+### Java Interopt
+To test if RSKafka can produce/consume records to/from the official Java client, you need to have Java installed and the
+`TEST_JAVA_INTEROPT=1` environment variable set.
+
 ### Fuzzing
 RSKafka offers fuzz targets for certain protocol parsing steps. To build them make sure you have [cargo-fuzz] installed.
 Select one of the following fuzzers:

src/protocol/record.rs

@@ -619,49 +619,39 @@ where
             }
             #[cfg(feature = "compression-snappy")]
             RecordBatchCompression::Snappy => {
-                use crate::protocol::vec_builder::DEFAULT_BLOCK_SIZE;
-                use snap::raw::{decompress_len, Decoder};
-
                 // Construct the input for the raw decoder.
                 let mut input = vec![];
                 reader.read_to_end(&mut input)?;
 
-                // The snappy compression used here is unframed aka "raw". So we first need to figure out the
-                // uncompressed length. See
-                //
-                // - https://github.com/edenhill/librdkafka/blob/2b76b65212e5efda213961d5f84e565038036270/src/rdkafka_msgset_reader.c#L345-L348
-                // - https://github.com/edenhill/librdkafka/blob/747f77c98fbddf7dc6508f76398e0fc9ee91450f/src/snappy.c#L779
-                let uncompressed_size = decompress_len(&input).unwrap();
-
-                // Decode snappy payload.
-                // The uncompressed length is unchecked and can be up to 2^32-1 bytes. To avoid a DDoS vector we try to
-                // limit it to a small size and if that fails we double that size;
-                let mut max_uncompressed_size = DEFAULT_BLOCK_SIZE;
-                let output = loop {
-                    let try_uncompressed_size = uncompressed_size.min(max_uncompressed_size);
-
-                    let mut decoder = Decoder::new();
-                    let mut output = vec![0; try_uncompressed_size];
-                    let actual_uncompressed_size = match decoder.decompress(&input, &mut output) {
-                        Ok(size) => size,
-                        Err(snap::Error::BufferTooSmall { .. })
-                            if max_uncompressed_size < uncompressed_size =>
-                        {
-                            // try larger buffer
-                            max_uncompressed_size *= 2;
-                            continue;
-                        }
-                        Err(e) => {
-                            return Err(ReadError::Malformed(Box::new(e)));
-                        }
-                    };
-                    if actual_uncompressed_size != uncompressed_size {
-                        return Err(ReadError::Malformed(
-                            "broken snappy data".to_string().into(),
-                        ));
+                const JAVA_MAGIC: &[u8] = &[0x82, b'S', b'N', b'A', b'P', b'P', b'Y', 0];
+
+                // There are "normal" compression libs, and there is Java
+                // See https://github.com/edenhill/librdkafka/blob/2b76b65212e5efda213961d5f84e565038036270/src/rdkafka_msgset_reader.c#L307-L318
+                let output = if input.starts_with(JAVA_MAGIC) {
+                    let mut cursor = Cursor::new(&input[JAVA_MAGIC.len()..]);
+
+                    let mut buf_version = [0u8; 4];
+                    cursor.read_exact(&mut buf_version)?;
+
+                    let mut buf_compatible = [0u8; 4];
+                    cursor.read_exact(&mut buf_compatible)?;
+
+                    let mut output = vec![];
+                    while cursor.position() < cursor.get_ref().len() as u64 {
+                        let mut buf_chunk_length = [0u8; 4];
+                        cursor.read_exact(&mut buf_chunk_length)?;
+                        let chunk_length = u32::from_be_bytes(buf_chunk_length) as usize;
+
+                        let mut chunk_data = vec![0u8; chunk_length];
+                        cursor.read_exact(&mut chunk_data)?;
+
+                        let mut buf = carefully_decompress_snappy(&chunk_data)?;
+                        output.append(&mut buf);
                     }
 
-                    break output;
+                    output
+                } else {
+                    carefully_decompress_snappy(&input)?
                 };
 
                 // Read uncompressed records.
@@ -889,6 +879,51 @@ where
     }
 }
 
+#[cfg(feature = "compression-snappy")]
+fn carefully_decompress_snappy(input: &[u8]) -> Result<Vec<u8>, ReadError> {
+    use crate::protocol::vec_builder::DEFAULT_BLOCK_SIZE;
+    use snap::raw::{decompress_len, Decoder};
+
+    // The snappy compression used here is unframed aka "raw". So we first need to figure out the
+    // uncompressed length. See
+    //
+    // - https://github.com/edenhill/librdkafka/blob/2b76b65212e5efda213961d5f84e565038036270/src/rdkafka_msgset_reader.c#L345-L348
+    // - https://github.com/edenhill/librdkafka/blob/747f77c98fbddf7dc6508f76398e0fc9ee91450f/src/snappy.c#L779
+    let uncompressed_size = decompress_len(input).map_err(|e| ReadError::Malformed(Box::new(e)))?;
+
+    // Decode snappy payload.
+    // The uncompressed length is unchecked and can be up to 2^32-1 bytes. To avoid a DDoS vector we try to
+    // limit it to a small size and if that fails we double that size;
+    let mut max_uncompressed_size = DEFAULT_BLOCK_SIZE;
+
+    loop {
+        let try_uncompressed_size = uncompressed_size.min(max_uncompressed_size);
+
+        let mut decoder = Decoder::new();
+        let mut output = vec![0; try_uncompressed_size];
+        let actual_uncompressed_size = match decoder.decompress(input, &mut output) {
+            Ok(size) => size,
+            Err(snap::Error::BufferTooSmall { .. })
+                if max_uncompressed_size < uncompressed_size =>
+            {
+                // try larger buffer
+                max_uncompressed_size *= 2;
+                continue;
+            }
+            Err(e) => {
+                return Err(ReadError::Malformed(Box::new(e)));
+            }
+        };
+        if actual_uncompressed_size != uncompressed_size {
+            return Err(ReadError::Malformed(
+                "broken snappy data".to_string().into(),
+            ));
+        }
+
+        break Ok(output);
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::io::Cursor;
@@ -1133,6 +1168,72 @@ mod tests {
         assert_eq!(actual2, expected);
     }
 
+    #[cfg(feature = "compression-snappy")]
+    #[test]
+    fn test_decode_fixture_snappy_java() {
+        // This data was obtained by watching Kafka returning a recording to rskafka that was produced by the official
+        // Java client.
+        let data = [
+            b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x8c\x00\x00\x00\x00".to_vec(),
+            b"\x02\x79\x1e\x2d\xce\x00\x02\x00\x00\x00\x01\x00\x00\x01\x7f\x07".to_vec(),
+            b"\x25\x7a\xb1\x00\x00\x01\x7f\x07\x25\x7a\xb1\xff\xff\xff\xff\xff".to_vec(),
+            b"\xff\xff\xff\xff\xff\xff\xff\xff\xff\x00\x00\x00\x02\x82\x53\x4e".to_vec(),
+            b"\x41\x50\x50\x59\x00\x00\x00\x00\x01\x00\x00\x00\x01\x00\x00\x00".to_vec(),
+            b"\x47\xff\x01\x1c\xfc\x01\x00\x00\x00\xc8\x01\x78\xfe\x01\x00\x8a".to_vec(),
+            b"\x01\x00\x64\x16\x68\x65\x6c\x6c\x6f\x20\x6b\x61\x66\x6b\x61\x02".to_vec(),
+            b"\x06\x66\x6f\x6f\x06\x62\x61\x72\xfa\x01\x00\x00\x02\xfe\x80\x00".to_vec(),
+            b"\x96\x80\x00\x4c\x14\x73\x6f\x6d\x65\x20\x76\x61\x6c\x75\x65\x02".to_vec(),
+            b"\x06\x66\x6f\x6f\x06\x62\x61\x72".to_vec(),
+        ]
+        .concat();
+
+        let actual = RecordBatch::read(&mut Cursor::new(data)).unwrap();
+        let expected = RecordBatch {
+            base_offset: 0,
+            partition_leader_epoch: 0,
+            last_offset_delta: 1,
+            first_timestamp: 1645092371121,
+            max_timestamp: 1645092371121,
+            producer_id: -1,
+            producer_epoch: -1,
+            base_sequence: -1,
+            records: ControlBatchOrRecords::Records(vec![
+                Record {
+                    timestamp_delta: 0,
+                    offset_delta: 0,
+                    key: Some(vec![b'x'; 100]),
+                    value: Some(b"hello kafka".to_vec()),
+                    headers: vec![RecordHeader {
+                        key: "foo".to_owned(),
+                        value: b"bar".to_vec(),
+                    }],
+                },
+                Record {
+                    timestamp_delta: 0,
+                    offset_delta: 1,
+                    key: Some(vec![b'x'; 100]),
+                    value: Some(b"some value".to_vec()),
+                    headers: vec![RecordHeader {
+                        key: "foo".to_owned(),
+                        value: b"bar".to_vec(),
+                    }],
+                },
+            ]),
+            compression: RecordBatchCompression::Snappy,
+            is_transactional: false,
+            timestamp_type: RecordBatchTimestampType::CreateTime,
+        };
+        assert_eq!(actual, expected);
+
+        let mut data2 = vec![];
+        actual.write(&mut data2).unwrap();
+
+        // don't compare if the data is equal because compression encoder might work slightly differently, use another
+        // roundtrip instead
+        let actual2 = RecordBatch::read(&mut Cursor::new(data2)).unwrap();
+        assert_eq!(actual2, expected);
+    }
+
     #[cfg(feature = "compression-zstd")]
     #[test]
     fn test_decode_fixture_zstd() {