DEV Community: Ivan Yurchenko

Embedding JVM in Rust

Ivan Yurchenko — Sun, 04 Jan 2026 14:40:16 +0000

Java ecosystem is rich and mature. I won't be surprised if some application may need to call a piece of Java code. There are several options to do this, starting from the basic multi-processing. In this short post, I want to tell about embedding Java virtual machine (JVM) directly into Rust programs with the jni crate.

In this post:

Embedding JVM into a Rust application.
Calling Java code from Rust and Rust code from Java.

The post assumes you're familiar with Java and Rust. You can find the code from this post in this repository on Github.

A toy example we will use may look very artificial, but this is to cover more features of the Java-Rust interaction. Here's our Java library code. We have three classes:

package me.ivanyu.java_library;

public class Calculator {
    public static void add(int a, int b, ResultCallback callback) {
        int sum = a + b;
        String message = "Result: " + sum;
        Result result = new Result(message);
        callback.onResult(result);
    }
}

// ...

@FunctionalInterface
public interface ResultCallback {
    void onResult(Result result);
}

// ...

public class Result {
    private final String message;

    public Result(final String message) {
        this.message = message;
    }

    public String getMessage() {
        return message;
    }
}

Calculator.add calculates the sum of two integers, makes a string of it and calls the callback with the result. Now we want to call this function from Rust code. We also want the callback to be a Rust function: Java → Rust → Java. We will use a Java mechanism called Java Native Interface (JNI) to achieve this.

First, we need to create a native Java method to which we can attach our native Rust code.

package me.ivanyu.java_library;

public class NativeCallback implements ResultCallback {
    @Override
    // Mind the `native` keyword.
    public native void onResult(Result result);
}

Now let's build a JAR of our library:

./gradlew clean jar

Coming to Rust, we need a crate called jni:

[dependencies]
jni = { version = "0.21.1", features = ["invocation"] }

Mind the invocation feature, it's needed to run a JVM from our Rust code.

Let's define our callback:

// Prevent symbol mangling
// https://doc.rust-lang.org/rustc/symbol-mangling/index.html
#[unsafe(no_mangle)]
// `extern "system"` ensures the correct calling convention for JNI.
// The name follows the convention from
// https://docs.oracle.com/en/java/javase/21/docs/specs/jni/design.html#resolving-native-method-names
// However, generally this is not needed because we use RegisterNatives function
// (through `jni` crate).
pub extern "system" fn Java_me_ivanyu_java_1library_NativeCallback_onResult(
    mut env: JNIEnv,
    _obj: JObject,
    result_obj: JObject,
) {
    // Call getMessage on the result.
    // The signature means
    // "no arguments, returns `java.lang.String`".
    // More details on signatures:
    // https://docs.oracle.com/en/java/javase/21/docs/specs/jni/types.html
    let message_jstring = env
        .call_method(result_obj, "getMessage", "()Ljava/lang/String;", &[])
        .expect("Failed to call getMessage")
        .l()  // unwrap to Object (implicitly check for null)
        .expect("getMessage returned null");
    // Convert a Java string to Rust string.
    let message_jstring = JString::from(message_jstring);
    let message: String = env
        .get_string(&message_jstring)
        .expect("Couldn't get Java string")
        .into();

    println!("{}", message);
}

Now let's run the JVM with our library in the class path and call the Calculator.add method. Make sure you've built the Java library first (the JAR file must exist at java/lib/build/libs/lib.jar):

fn main() {
    // Add the Java library to the class path.
    let jar_path = Path::new("java/lib/build/libs/lib.jar");
    let classpath_option = format!("-Djava.class.path={}", jar_path.display());
    let jvm_args = jni::InitArgsBuilder::new()
        .version(jni::JNIVersion::V8)
        .option(&classpath_option)
        .build()
        .expect("Failed to create JVM args");
    // Launch the JVM.
    let jvm = JavaVM::new(jvm_args).expect("Failed to create JVM");

    // The JNI interface pointer (JNIEnv) is valid only in the current thread. We need to attach it.
    // See more:
    // https://docs.oracle.com/en/java/javase/21/docs/specs/jni/invocation.html#attaching-to-the-vm
    // Detachment happens automatically when `env` is deleted.
    let mut env = jvm
        .attach_current_thread()
        .expect("Failed to attach current thread");

    // Find the native callback class and register the native method for it.
    let native_callback_class = env
        .find_class("me/ivanyu/java_library/NativeCallback")
        .expect("Failed to find NativeCallback class");
    // The signature means
    // "one argument of the class `me.ivanyu.java_library.Result`, returning `void`".
    // More details on signatures:
    // https://docs.oracle.com/en/java/javase/21/docs/specs/jni/types.html
    let native_methods = [jni::NativeMethod {
        name: jni::strings::JNIString::from("onResult"),
        sig: jni::strings::JNIString::from("(Lme/ivanyu/java_library/Result;)V"),
        fn_ptr: Java_me_ivanyu_java_1library_NativeCallback_onResult as *mut c_void,
    }];
    env.register_native_methods(&native_callback_class, &native_methods)
        .expect("Failed to register native methods");

    // Create an instance of NativeCallback
    let callback_obj = env
        // Our native callback code.
        .alloc_object(&native_callback_class)
        .expect("Failed to allocate NativeCallback object");

    // Find the Calculator class and call the `add` method (static).
    let calculator_class = env
        .find_class("me/ivanyu/java_library/Calculator")
        .expect("Failed to find Calculator class");
    env.call_static_method(
        &calculator_class,
        "add",
        "(IILme/ivanyu/java_library/ResultCallback;)V",
        &[
            jni::objects::JValue::Int(3),
            jni::objects::JValue::Int(5),
            jni::objects::JValue::Object(&callback_obj),
        ],
    )
    .expect("Failed to call add");
}

That's it, let's run:

cargo run 
Result: 8

There is another similar crate, Duchess, which is more high-level and focused on ergonomics. It uses marcos to reflect Java classes into Rust types and call methods as if they were native Rust methods. This may be a topic for another post.

Proptest: property testing in Rust

Ivan Yurchenko — Fri, 27 Dec 2024 17:04:17 +0000

You probably heard about property testing (AKA property-based testing), a testing technique where test inputs are randomly generated in great quantity and particular desired properties of the code under test are checked. It's very practical and I'm a big proponent of using it where it makes sense.

In this post, I will tell you how I used property testing with the Proptest library in Rust to ensure the correctness of a bunch of generated serialization/deserialization code for the Apache Kafka protocol.

In this post:

A real-life example of using property testing in Rust with Proptest.
A step-by-step explanation.

The post assumes you're familiar with Rust. However, to follow it you don't need any knowledge about Kafka or its protocol. I will explain the problem I was working on briefly, but will not go into much detail.

Introduction

Apache Kafka has a sizeable custom binary protocol that is versioned, with various data types, optional fields, etc. Unfortunately, it doesn't use a well-known serialization format like Protobuf. The protocol message schema is described in JSON. The actual Java code that does serialization and deserialization is generated from this description¹.

I created a library kafka_wire_protocol. It's a generated de-/serialization code for the Kafka protocol, but for Rust (and in the future, for Go). The correctness of the generated code is paramount. I applied many testing techniques to it: quick unit tests, integration tests against a real Kafka instance, fuzzying. And also kafka_wire_protocol is a good example of software that can benefit from property testing. About this is going to be this post.

You don't need to familiarize yourself with kafka_wire_protocol to understand what will be discussed. I will give examples from the code, but they will be self-contained. However, if you're interested, feel free to check out this repo and follow the instruction in README to run the code generator, compile, and run the tests.

Property testing

If we quote Wikipedia:

Property testing is a testing technique where, instead of asserting that specific inputs produce specific expected outputs, the practitioner randomly generates many inputs, runs the program on all of them, and asserts the truth of some "property" that should be true for every pair of input and output.

The programmer writes the test code similar to unit tests, i.e. an input, some operation(s) on this input, and finally some checks on the output. The set of checks represents the properties we're testing in code. Normally, some property testing library is used to support this. The library does two main things:

It generates many inputs randomly with the guidance from programmer. It should be convenient for the human to set up the strategy for generation, it should be mostly automatic and easily composable from building blocks.
Once a violation of a property under test is detected, the library shrinks the input, that is, minifies it to make it more comprehensible to people.

For example, every output from a serialization function should be accepted by the corresponding deserialization function

This sounds quite like what we want to achieve here! Property testing is particularly good for "paired" functions, like serialization/deserialization, encryption/decryption, compression/decompression, and so on, because they are so easy to check against each other.

Property testing was invented (to the best of my knowledge) by Koen Claessen and John Hughes in the library QuickCheck back in 1999. If you're interested, you can read the story of QuickCheck here.

Originally made for Haskell, QuickCheck inspired many similar libraries for numerous programming platforms.

Proptest

Proptest, inspired by Hypothesis (a Python library), is a property testing library for Rust. I used Proptest for kafka_wire_protocol.

Step by step towards property tests

Let's manually re-create the generated code with property tests step by step. We will use MetadataRequest version 12. If you're interested in the final generated variant of the code in the following sections, check out this file. Meanwhile, I skip all the cfg(test) attributes here to not clutter the code.

Step 1. Simplified generated code

The root struct is called MetadataRequest, which contains a vector of another struct MetadataRequestTopic. The implementations of Readable and Writable are generated, so feel free to ignore what's inside them.

use std::io::{Read, Result, Write};

use uuid::Uuid;

use crate::arrays::{read_nullable_array, write_nullable_array};
use crate::readable_writable::{Readable, Writable};
use crate::tagged_fields::{read_tagged_fields, write_tagged_fields, RawTaggedField};

#[derive(Clone, PartialEq)]
pub struct MetadataRequest {
    pub topics: Option<Vec<MetadataRequestTopic>>,
    pub allow_auto_topic_creation: bool,
    pub include_topic_authorized_operations: bool,
    pub _unknown_tagged_fields: Vec<RawTaggedField>,
}

/* Generated code */
impl Readable for MetadataRequest {
    fn read(#[allow(unused)] input: &mut impl Read) -> Result<Self> {
        let topics = read_nullable_array::<MetadataRequestTopic>(input, "topics", true)?;
        let allow_auto_topic_creation = bool::read(input)?;
        let include_topic_authorized_operations = bool::read(input)?;
        let tagged_fields_callback = |tag: i32, _: &[u8]| {
            match tag {
                _ => Ok(false)
            }
        };
        let _unknown_tagged_fields = read_tagged_fields(input, tagged_fields_callback)?;
        Ok(MetadataRequest {
            topics, allow_auto_topic_creation, include_topic_authorized_operations, _unknown_tagged_fields
        })
    }
}

/* Generated code */
impl Writable for MetadataRequest {
    fn write(&self, #[allow(unused)] output: &mut impl Write) -> Result<()> {
        write_nullable_array(output, "topics", self.topics.as_deref(), true)?;
        self.allow_auto_topic_creation.write(output)?;
        self.include_topic_authorized_operations.write(output)?;
        write_tagged_fields(output, &[], &self._unknown_tagged_fields)?;
        Ok(())
    }
}

#[derive(Clone, PartialEq)]
pub struct MetadataRequestTopic {
    pub topic_id: Uuid,
    pub name: Option<String>,
    pub _unknown_tagged_fields: Vec<RawTaggedField>,
}

/* Generated code */
impl Readable for MetadataRequestTopic {
    fn read(#[allow(unused)] input: &mut impl Read) -> Result<Self> {
        let topic_id = Uuid::read(input)?;
        let name = Option::<String>::read_ext(input, "name", true)?;
        let tagged_fields_callback = |tag: i32, _: &[u8]| {
            match tag {
                _ => Ok(false)
            }
        };
        let _unknown_tagged_fields = read_tagged_fields(input, tagged_fields_callback)?;
        Ok(MetadataRequestTopic {
            topic_id, name, _unknown_tagged_fields
        })
    }
}

/* Generated code */
impl Writable for MetadataRequestTopic {
    fn write(&self, #[allow(unused)] output: &mut impl Write) -> Result<()> {
        self.topic_id.write(output)?;
        self.name.write_ext(output, "name", true)?;
        write_tagged_fields(output, &[], &self._unknown_tagged_fields)?;
        Ok(())
    }
}

If we ignore the meaning of these fields, this is quite trivial a situation of a simple struct containing a bunch of instances of another struct + some extra data. You can find this in every other program.

Step 2. Property test for serialization/deserialization

First, let's add the proptest crate to the project. Then, write code like this:

#[cfg(test)]
mod tests {
    use std::io::{Cursor, Seek, SeekFrom};

    use proptest::prelude::*;

    use super::*;

    proptest! {
        #[test]
        fn test_serde(data: MetadataRequest) {
            // Serialize.
            let mut cur = Cursor::new(Vec::<u8>::new());
            data.write(&mut cur).unwrap();

            // Deserialize.
            cur.seek(SeekFrom::Start(0)).unwrap();
            let data_read = MetadataRequest::read(&mut cur).unwrap();

            // Compare.
            prop_assert_eq!(data_read, data.clone());
        }
    }
}

As you see, we don't explicitly create instances of MetadataRequest. Due to the proptest! macro, the Proptest does this automatically for us.

However, if we try to compile this, it won't:

error[E0277]: the trait bound `v12::MetadataRequest: Arbitrary` is not satisfied
   --> src/schema/metadata_request/v12.rs:90:29
    |
90  |         fn test_serde(data: MetadataRequest) {
    |                             ^^^^^^^^^^^^^^^ the trait `Arbitrary` is not implemented for `v12::MetadataRequest`

This means that the macro is expecting an implementation of Arbitrary to be available for the structure MetadataRequest for the magic to work.

Let's understand the central concept of the Proptest library (and many other property testing libraries), which is Strategy. Quoting the documentation:

A strategy defines two things:

How to generate random values of a particular type from a random number generator.

How to “shrink” such values into “simpler” forms.

Arbitrary determines a canonical Strategy for the implementing type. To generate test instances, Proptest requires the type to implement the Arbitrary trait (there are other ways to tell Proptest what Strategy to use, too).

There are many strategies for fundamental types provided by the library. Let's implement (or rather, derive) the strategy for our structs.

Step 3. Deriving `Arbitrary`

Implementing Arbitrary manually for each struct we may be interested in would take a lot of boring and error-prone work. Luckily, the automatic generation of simple strategies for structs, enums is straightforward and can be done automatically with the proptest-derive crate. It provides the #[derive(Arbitrary)] macro, which we're going to use here. We will also need Debug (because Arbitrary needs it), and Clone and PartialEq (for the test code), which also could be derived.

#[derive(Arbitrary, Debug, Clone, PartialEq)]
pub struct MetadataRequest {
...

Since derivation is recursive, the same is needed for MetadataRequestTopic used by MetadataRequest:

#[derive(Arbitrary, Debug, Clone, PartialEq)]
pub struct MetadataRequestTopic {
...

However, this still doesn't compile:

error[E0277]: the trait bound `uuid::Uuid: types::_::_proptest::arbitrary::Arbitrary` is not satisfied
 --> src/schema/metadata_request/v12.rs:9:10
  |
9 | #[derive(Arbitrary, Debug)]
  |          ^^^^^^^^^ the trait `types::_::_proptest::arbitrary::Arbitrary` is not implemented for `uuid::Uuid`

MetadataRequestTopic uses Uuid inside, which doesn't implement Arbitrary. Moreover, Arbitrary and Uuid are defined in different crates. Rust prohibits implementing the former for the latter following the "orphan rule". A workaround may be to create a new wrapper type. However, this is a public-facing API of kafka_wire_protocol and it didn't sound great to me to expose a newtype for UUIDs there.

Proptest offers another workaround.

Step 4. Custom strategy for `Uuid`

Apart from the default strategy provided by Arbitrary, one can define as many custom strategies as needed and instruct proptest-derive to use it with annotations. Here you can find some info on how to create custom strategies.

What's a random UUID? It's random 128 bits. Proptest has two things:

The out-of-the-box strategy for generating 128-bit numbers.
A way to transform strategies to produce new strategies.

Let's combine these two:

pub(crate) fn uuid() -> impl Strategy<Value=Uuid> {
    any::<u128>().prop_map(Uuid::from_u128)
}

Now let's ask Proptest to use this strategy when deriving Arbitrary for MetadataRequestTopic. First, import the strategy if it's defined in another module:

use crate::test_utils::proptest_strategies;

Then, annotate the field in the MetadataRequestTopic struct:

#[derive(Arbitrary, Debug, Clone, PartialEq)]
pub struct MetadataRequestTopic {
    #[proptest(strategy = "proptest_strategies::uuid()")]
    pub topic_id: Uuid,
    pub name: Option<String>,
    pub _unknown_tagged_fields: Vec<RawTaggedField>,
}

Step 5. First run and first failure

The code finally compiles, let's run the tests:

Test failed: called `Result::unwrap()` on an `Err` value: Custom { kind: Other, error: "Invalid raw tag field list: tag -1 comes after tag -1, but is not higher than it." }.
minimal failing input: data = MetadataRequest {
    topics: None,
    allow_auto_topic_creation: false,
    include_topic_authorized_operations: false,
    _unknown_tagged_fields: [
        RawTaggedField {
            tag: -1,
            data: [],
        },
        RawTaggedField {
            tag: 0,
            data: [],
        },
    ],
}
    successes: 0
    local rejects: 0
    global rejects: 0
thread 'schema::metadata_request::v12::tests::test_serde' panicked at src/schema/metadata_request/v12.rs:97:34:
called `Result::unwrap()` on an `Err` value: Custom { kind: Other, error: "Invalid raw tag field list: tag -358390220 comes after tag -1, but is not higher than it." }

(Your run may have different tag values or number of tagged fields, but the idea will be the same.)

The run failed. The error is Invalid raw tag field list: tag -1 comes after tag -1, but is not higher than it. It's the details of the implementation (you can read the code here if you're interested), but the high-level idea is that it's not allowed to have two tagged fields with non-increasing tags. So, the serde code may be correct, but the derived Proptest strategy isn't. Let's fix this issue.

Step 6. Custom strategy for `Vec<RawTaggedField>`

There are multiple types of strategies to employ for generating this _unknown_tagged_fields field. However, a simple one will suffice. Defined in human language, it would read:

Either no unknown tagged fields, or a single field tagged with 999 and random bytes.

Let's define it:

pub(crate) fn unknown_tagged_fields() -> impl Strategy<Value=Vec<RawTaggedField>> {
    prop_oneof![
        Just(Vec::<RawTaggedField>::new()),
        bytes().prop_map(|data| vec![RawTaggedField{ tag: 999, data }]),
    ]
}

The prop_oneof! macro tells Proptest to select one of its elements, which are Strategy themselves. It's like an if of switch statement in the randomized property testing world. For example, check out how it can be used for generating enums. prop_oneof! supports assigning weights to variants in case you need some of them to appear more frequently than others.

Just is a very simple strategy that always produces the provided value. In this case, the strategy gives out an empty vector of RawTaggedField.

In the second arm of prop_oneof! there's the strategy for creating a single-element vector of RawTaggedField. It's created by transformation from our yet-to-be-written random byte vector strategy
bytes() into a single element vector strategy where the bytes represent the tagged field data.

Let's now define bytes(). Proptest offers an out-of-the-box strategy for Vec<u8>. However, it works with an arbitrary length. Huge values would not bring any new information to the tests, only slow them down and increase the resource consumption. So let's limit the length of the output.

pub(crate) fn bytes() -> impl Strategy<Value=Vec<u8>> {
    collection::vec(prop::num::u8::ANY, collection::size_range(0..10))
}

collection::vec is Proptest's way to define a vector strategy based on an element strategy. In the second parameter, we define the target range of the vector.

Having created the unknown_tagged_fields strategy, let's apply it to our structs:

#[derive(Arbitrary, Debug, Clone, PartialEq)]
pub struct MetadataRequest {
    ...
    #[proptest(strategy = "proptest_strategies::unknown_tagged_fields()")]
    pub _unknown_tagged_fields: Vec<RawTaggedField>,
}

#[derive(Arbitrary, Debug, Clone, PartialEq)]
pub struct MetadataRequestTopic {
    ...
    #[proptest(strategy = "proptest_strategies::unknown_tagged_fields()")]
    pub _unknown_tagged_fields: Vec<RawTaggedField>,
}

Now when we run the test, it will finally pass successfully.

Conclusion

There's more to Proptest that I wrote in this post (but not much more), please check the documentation. But I hope this practical introduction will help you get started with the library in your projects. Also, check out similar libraries for different languages like the mentioned Hypothesis for Python or jqwik for Java.

Foot note: testing against real Java Kafka code

With property testing done, we're more or less sure that our serialization and deserialization work correctly together. But what if they contain a dual error that makes them correct with respect to each other, but incorrect for the real Kafka serialization/deserialization, which we consider the gold standard and source of truth? There's only one way to be sure about that, and that is to test against the Java code.

In my day job, we developed an open source library Kio. It essentially does what kafka_wire_protocol does but in Python. Kio has a component named (very straightforwardly) Java Tester, implemneted by yours truly some time ago. Its point was simple yet powerful: We generate random objects like in our serde tests. But instead of just checking the serialization-deserialization pair with our generated code, why don't we also ask the real Java Kafka serde code to do the same and compare the results? Java Tester did exactly that: it serialized the object in binary form and also in JSON and sends these two representations to the special Java tester process. This process, in turn, reconstructed the object based on its JSON representation and then serialized it in the binary form itself. After that, it compared what the external code produced with its own result. The expectation is byte-to-byte equality, otherwise an error is reported.

The Java Tester approach is what was needed for kafka_wire_protocol. So I copied it from Kio with a few changes. It's pretty sizeable (due to the reconstruction logic), you can check it out in the repo. I'm not going to explain its internal details, because it'd be mostly off-topic.

See my recent post on the Kafka protocol. ↩

Kafka protocol practical guide

Ivan Yurchenko — Thu, 26 Dec 2024 05:24:29 +0000

I worked with the Apache Kafka protocol on the low level quite a bit. It wasn't easy to start doing this following the official guide only and I read the code a lot. With this post, I want to give you a head start by guiding you step by step from primitive values to meaningful requests.

In this post:

Explore the Kafka protocol code and the protocol in action with Wireshark.
Learn how to read and write primitive values.
Combine primitives to perform meaningful requests.

We will use Python as the programming language. However, the code will be zero-dependency and easily portable to the language of your choice.

Intro

Apache Kafka has a custom binary protocol that is versioned, with various data types, optional fields, etc. Unfortunately, it doesn't use a well-known serialization format like Protobuf. The protocol message schema is described in JSON. The actual Java code that does serialization and deserialization is generated from this description.

When you're in the Java world, you can use the official client library. But if you're using another platform, you're relying on third-party implementations. They exist, but they're focusing mostly on the producer and consumer, rarely on some aspects of the admin client. If you need to do something else, you're on your own.

This post will help you start hacking on the Kafka protocol. (If you are looking for a ready Python (de-)serialization library for the Kafka protocol, check out Kio¹. For Rust, have a look at the library I'm working on.)

You can find the code from this post and some more like tests in this repository on Github.

Protocol overview

You can find the official protocol description on this page. I encourage you to familiarize yourself with it, at least read the "Preliminaries" and "The Protocol" sections.

Here are some highlights. The Kafka protocol is a TCP-based binary request-response protocol:

TCP-based: the Kafka broker listens at ports on the TCP stack (which gives some benefits like ordering guarantees).
Binary: messages are encoded in binary form and require special serialization and deserialization according to predefined schemas.
Request-response: exchanges are initiated by the client, the server is passive and only replies to requests.

Each API message type consists of the request and response pair and is identified by a numeric value called the API key. For example, Produce and Fetch, the most characteristic Kafka RPCs, have API keys 0 and 1 correspondingly. Nowadays, there are close to 90 API message types (some of them are inter-broker, not client-broker).

Requests and responses are described by versioned schemas. Versioning allows for protocol evolution, for example, adding or removing fields or changing their data type.

First steps

Here are some things you can do to start working with the Kafka protocol.

Learn Kafka protocol code

The Kafka code is the source of truth (practically) about the protocol. Check out the Kafka code from Github and switch to the release you're interested in (e.g. 3.8.0):

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

You can find the API message definitions in JSON in clients/src/main/resources/common/message. Each JSON file contains the definition of one message² type with all its versions. clients/src/main/resources/common/message/README.md gives a good overview of the schema definition format. Pay attention to stuff like default values, flexible versions, and tagged fields.

Apart from the concrete API message types you're interested in, have a look at clients/src/main/resources/common/message/RequestHeader.json and ResponseHeader.json, which describe headers used in each request-response exchange.

Let's run the code generator:

./gradlew processMessages

Now you can find the generated classes in clients/src/generated/java/org/apache/kafka/common/message.

Have a look at clients/src/generated/java/org/apache/kafka/common/message/ApiMessageType.java. This utility:

describes the whole set of existing API message types along with their schemas and versions;
maps API message versions to request and response header versions in the requestHeaderVersion and responseHeaderVersion functions.

Other files are one-to-one generated from the corresponding schema JSONs (sometimes with the Data postfix, it's a compatibility thing). In these files you'll find:

The versioned schema definitions SCHEMA_0, SCHEMA_1, etc. Sometimes schemas stay the same between versions. This is normal and means only the request-response counterpart changed.
read and write methods where you can find the ground truth for the protocol serialization and deserialization.

Pay attention to the inner classes as well as they represent the complex structure of the message.

Run Kafka in Docker

Running Kafka in Docker is a convenient way to get a broker running to test the protocol or capture the network exchange. Since version 3.7.0, the Kafka team builds official Docker images, which you can run as:

docker run --rm -ti -p 9092:9092 apache/kafka:3.8.0

If you're interested in older versions, search Docker Hub for other images. However, this may be not needed considering the Kafka protocol is backward and forward compatible: new brokers will recognize the old protocol versions just fine and the old clients can communicate with newer brokers.

If you read this, you probably already have the Kafka command line tools on your machine, but just in case, you can also run them in Docker. For example, run this to create a topic:

docker run --rm -ti --net=host apache/kafka:3.8.0 \
  /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

Inspect protocol with Wireshark

Having familiarized ourselves with the Kafka code, let's look at the protocol in action. Wireshark is a widely used tool for such inspections. It can dissect the Kafka protocol (and supports the latest versions if your version is fresh enough).

I built Wireshark from sources of version 4.5.0, because my operating system package is old and not able to dissect Kafka protocol with new versions. Wireshark 4.5.0 should mostly support Kafka 3.7 protocol versions. However, you can try the available version and see how it works for you.

Let's run Wireshark on the loopback interface with the port 9092 capture filter (1) and the kafka display filter (2):

Create a topic and see what Wireshark shows us:

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

The display filter removes everything irrelevant and leaves only Kafka requests and responses. As Wireshark understands most message versions in the protocol (depending on the Wireshark version, of course), you can conveniently look into the structure of each message. Wireshark will also show the corresponding bytes.

Wireshark is a great debugging tool that could help you understand how the protocol works in a particular case and what's wrong with your implementation.

Reading and writing primitive values

The protocol defines a number of primitive types, which full description you can find here. Let's implement the read and write code for them. You can find all functions in this file, also check out the corresponding test file.

Fixed length integer values: `INT8`, `INT16`, `INT32`, `INT64`, and `UINT16`

These are integer numbers with known fixed length: 1, 2, 4, or 8 bytes. Naturally, you can find such fields a lot throughout the protocol. In this class you may see how (trivially) their reading and writing are implemented in Kafka.

Let's first define the function for reading an exact number of bytes from a buffer³:

def read_exact(buffer: BinaryIO, num_bytes: int) -> bytes:
    value = buffer.read(num_bytes)
    if len(value) != num_bytes:
        raise ValueError(f"Buffer underflow: expected {num_bytes}, got {len(value)}")
    return value

The BinaryIO type hint in Python represents an object from which bytes can be read and to which they can be written. It has methods like read, write, tell (for getting the current position), seek (for changing the position).

Now we can implement reading INT8:

def read_int8(buffer: BinaryIO) -> int:
    return int.from_bytes(read_exact(buffer, 1), byteorder="big", signed=True)

Kafka uses the big-endian (AKA network) byte ordering, hence byteorder="big".

Now writing:

def write_int8(value: int, buffer: BinaryIO) -> None:
    if -(2**7) <= value <= 2**7 - 1:
        buffer.write(value.to_bytes(1, byteorder="big", signed=True))
    else:
        raise ValueError(f"Value {value} is out of range for INT8")

I won't repeat this for INT16, INT32, and INT64: the only significant differences are the number of bytes (2, 4, and 8 correspondingly) and checked ranges ([-(2**15), 2**15 - 1], [-(2**31), 2**31 - 1], and [-(2**63), 2**63 - 1] correspondingly).

UINT16 is similar to INT16:

def read_uint16(buffer: BinaryIO) -> int:
    return int.from_bytes(read_exact(buffer, 2), byteorder="big", signed=False)


def write_uint16(value: int, buffer: BinaryIO) -> None:
    if 0 <= value <= 2**16 - 1:
        buffer.write(value.to_bytes(2, byteorder="big", signed=False))
    else:
        raise ValueError(f"Value {value} is out of range for UINT16")

Note the signed=False here.

`BOOLEAN`

BOOLEAN is essentially INT8 with extra logic: == 0 means false, != 0 means true.

def read_boolean(buffer: BinaryIO) -> bool:
    return read_int8(buffer) != 0


def write_boolean(value: bool, buffer: BinaryIO) -> None:
    write_int8(1 if value is True else 0, buffer)

You can see an example of BOOLEAN in the allowAutoTopicCreation field of the MetadataRequestData generated class.

`FLOAT64`

FLOAT64 is a double-precision 64-bit IEEE 754 value. Python doesn't have to_bytes and from_bytes for float like it has for int. So instead we will use the struct module from the standard library.

def read_float64(buffer: BinaryIO) -> float:
    return struct.unpack(">d", read_exact(buffer, 8))[0]


def write_float64(value: float, buffer: BinaryIO) -> None:
    buffer.write(struct.pack(">d", value))

>d means "a double value in the big-endian byte order".

`UNSIGNED_VARINT`: Variable length integer values

Variable length integers are an approach that allows using of fewer bits per value when values are small. Kafka uses the varint approach from Protocol Buffers. The idea is simple:

Each byte in the varint has a continuation bit that indicates if the byte that follows it is part of the varint. This is the most significant bit (MSB) of the byte (sometimes also called the sign bit). The lower 7 bits are a payload; the resulting integer is built by appending together the 7-bit payloads of its constituent bytes.

You can check the Protobuf specification and Kafka implementation (read, write) for details.

This type isn't used in the protocol fields per se, but it's used for compact collections described below.

Let's implement it. For confidence, we get some examples directly from the source of truth, Kafka's ByteUtils class:

public class VarintCases {
    public static void main(String[] args) {
        for (Integer value : new int[]{0, 1, 127, 128, 129, 256, 1024, 100500, 9999999, Integer.MAX_VALUE}) {
            ByteBuffer buffer = ByteBuffer.allocate(5);
            ByteUtils.writeUnsignedVarint(value, buffer);
            System.out.printf("%10d\t", value);
            byte[] array = buffer.array();
            for (int i = 0; i < buffer.position(); i++) {
                System.out.printf("%s ", formatByte(array[i]));
            }
            System.out.println();
        }
    }

    private static String formatByte(byte b) {
        String result = String.format("%8s", Integer.toBinaryString(b))
                .replace(' ', '0');
        result = result.substring(result.length() - 8);
        result = "[" + result.charAt(0) + "|" + result.substring(1) + "]";
        return result;
    }
}

Running this, we'll get:

         0  [0|0000000] 
         1  [0|0000001] 
       127  [0|1111111] 
       128  [1|0000000] [0|0000001] 
       129  [1|0000001] [0|0000001] 
       256  [1|0000000] [0|0000010] 
      1024  [1|0000000] [0|0001000] 
    100500  [1|0010100] [1|0010001] [0|0000110] 
   9999999  [1|1111111] [1|0101100] [1|1100010] [0|0000100] 
2147483647  [1|1111111] [1|1111111] [1|1111111] [1|1111111] [0|0000111]

Let's implement this in probably not the most performant, but a straightforward way:

def read_unsigned_varint(buffer: BinaryIO) -> int:
    result = 0
    # Go by 7 bit steps.
    for offset in [0, 7, 14, 21, 28]:
        byte = int.from_bytes(read_exact(buffer, 1), byteorder="big", signed=False)

        # Concat the payload, 7 lower bits, to the result.
        payload_bits = byte & 0b111_1111
        result |= payload_bits << offset

        # This is the last byte if its most significant bit is 0.
        if byte & 0b1000_0000 == 0:
            return result
    else:
        raise ValueError(
            "Varint is too long, most significant bit in 5th byte is set"
        )


def write_unsigned_varint(value: int, buffer: BinaryIO) -> None:
    if value < 0 or value > 2**31 - 1:
        raise ValueError(f"Value {value} is out of range for UNSIGNED VARINT")

    written = False  # has at least one byte been written?
    while not written or value > 0:
        byte_to_write = value & 0b111_1111  # 7 lower bits
        value = value >> 7
        # Add the bit that signifies that more is to come.
        if value > 0:
            byte_to_write |= 0b1000_0000
        buffer.write(byte_to_write.to_bytes(1, byteorder="big", signed=False))
        written = True

`UUID`

UUID are 128-bit values used for uniquely identifying entities. For example, they are used to pass topic IDs in CreateTopicsResponse.

You can see how they are read and written in the Kafka code. It's simple to reproduce:

def read_uuid(buffer: BinaryIO) -> UUID | None:
    byte_value: bytes = read_exact(buffer, 16)
    if byte_value == UUID_ZERO.bytes:
        return None
    else:
        return UUID(bytes=byte_value)


def write_uuid(value: UUID | None, buffer: BinaryIO) -> None:
    if value is None:
        buffer.write(UUID_ZERO.bytes)
    else:
        buffer.write(value.bytes)

Note that Kafka treats null/None as zero UUID, so we're doing the same here.

Strings

The Kafka protocol has 4 types of strings:

	compact	non-compact
nullable	`COMPACT_NULLABLE_STRING`	`NULLABLE_STRING`
non-nullable	`COMPACT_STRING`	`STRING`

Compactness indicates whether the string length is encoded with INT16 or with UNSIGNED_VARINT. It depends on the message version (it was introduced around 2017). Nullability is whether the value can be null. It depends on the message purpose and the version as well (sometimes string fields become optional during the protocol evolution).

Strings are ubiquitous in the protocol. For example, see the field name in the generated class MetadataRequestData.MetadataRequestTopic.

Strings are encoded pretty straightforwardly: first goes the length and then comes the UTF-8 encoded body. The maximum allowed length is 32767 bytes. Null strings have the length of -1 and obviously no body.

As the only difference between compact and non-compact is how the string length is encoded, we can have one function for both modes.

Let's start with reading and writing nullable strings:

def read_string_length(buffer: BinaryIO, compact: bool) -> int:
    # In the compact variant, stored lengths are increased by 1
    # to preserve unsignedness.
    length: int
    if compact:
        length = read_unsigned_varint(buffer) - 1
    else:
        length = read_int16(buffer)
    if length < -1 or length > 2**15 - 1:
        raise ValueError(f"string has invalid length {length}")
    return length


def read_nullable_string(buffer: BinaryIO, compact: bool) -> str | None:
    length = read_string_length(buffer, compact)
    if length == -1:
        return None
    else:
        return read_exact(buffer, length).decode(encoding="utf-8")


def write_string_length(length: int, buffer: BinaryIO, compact: bool) -> None:
    if length > 2**15 - 1:
        raise ValueError(f"string has invalid length {length}")

    # In the compact variant, stored lengths are increased by 1
    # to preserve unsignedness.
    if compact:
        write_unsigned_varint(length + 1, buffer)
    else:
        write_int16(length, buffer)


def write_nullable_string(value: str | None, buffer: BinaryIO, compact: bool) -> None:
    if value is None:
        write_string_length(-1, buffer, compact)
    else:
        value_b = value.encode(encoding="utf-8")
        write_string_length(len(value_b), buffer, compact)
        buffer.write(value_b)

Non-nullable string functions can be built on top of these:

def read_string(buffer: BinaryIO, compact: bool) -> str:
    result = read_nullable_string(buffer, compact)
    if result is None:
        raise ValueError("Non-nullable field was serialized as null")
    return result


def write_string(value: str, buffer: BinaryIO, compact: bool) -> None:
    write_nullable_string(value, buffer, compact)

Byte arrays

Byte arrays are very similar to strings. They have the same potential nullability and compactness:

	compact	non-compact
nullable	`COMPACT_NULLABLE_BYTES`	`NULLABLE_BYTES`
non-nullable	`COMPACT_BYTES`	`BYTES`

They are also encoded in the same way: length + body. Naturally, the body is not treated as an UTF-8 string, but as an opaque byte array. The max length of a byte array is 2147483647;

You can find an example of bytes in the field metadata in the generated class JoinGroupRequestData.JoinGroupRequestProtocol.

def read_array_length(buffer: BinaryIO, compact: bool) -> int:
    # In the compact variant, stored lengths are increased by 1
    # to preserve unsignedness.
    if compact:
        return read_unsigned_varint(buffer) - 1
    else:
        return read_int32(buffer)


def read_nullable_bytes(buffer: BinaryIO, compact: bool) -> bytes | None:
    length = read_array_length(buffer, compact)
    if length < -1 or length > 2**31 - 1:
        raise ValueError(f"bytes has invalid length {length}")

    if length == -1:
        return None
    else:
        return read_exact(buffer, length)


def write_array_length(length: int, buffer: BinaryIO, compact: bool) -> None:
    if length > 2**31 - 1:
        raise ValueError(f"bytes has invalid length {length}")

    # In the compact variant, stored lengths are increased by 1
    # to preserve unsignedness.
    if compact:
        write_unsigned_varint(length + 1, buffer)
    else:
        write_int32(length, buffer)


def write_nullable_bytes(value: bytes | None, buffer: BinaryIO, compact: bool) -> None:
    if value is None:
        write_array_length(-1, buffer, compact)
    else:
        write_array_length(len(value), buffer, compact)
        buffer.write(value)

As you can see, the difference between these functions and the corresponding functions for strings is small.

Other arrays

The protocol supports arrays of types other than bytes: strings, numbers, structs (but not nested arrays): ARRAY and COMPACT_ARRAY. Compactness is the same as in byte arrays and strings.

Nullability is not explicitly mentioned in the protocol specification for some reason. However, arrays can be nullable. This is controlled by nullableVersions in the schema definitions, like here.

Considering we already implemented read_array_length and write_array_length, let's implement the reader and writer functions:

def read_array(
    read_element: Callable[[BinaryIO], T], buffer: BinaryIO, compact: bool
) -> list[T]:
    result = read_nullable_array(read_element, buffer, compact)
    if result is None:
        raise ValueError("Non-nullable field was serialized as null")
    return result


def write_array(
    array: list[T],
    write_element: Callable[[T, BinaryIO], None],
    buffer: BinaryIO,
    compact: bool,
) -> None:
    write_nullable_array(array, write_element, buffer, compact)


def read_nullable_array(
    read_element: Callable[[BinaryIO], T], buffer: BinaryIO, compact: bool
) -> list[T] | None:
    length = read_array_length(buffer, compact)
    if length == -1:
        return None
    else:
        array = []
        for _ in range(length):
            array.append(read_element(buffer))
        return array


def write_nullable_array(
    array: list[T] | None,
    write_element: Callable[[T, BinaryIO], None],
    buffer: BinaryIO,
    compact: bool,
) -> None:
    if array is None:
        write_array_length(-1, buffer, compact)
    else:
        write_array_length(len(array), buffer, compact)
        for el in array:
            write_element(el, buffer)

`RECORDS`

RECORDS encode Kafka records. The structure is pretty complex and I'm not going to describe it in this guide (however, please let me know in the comments 👇️ if you would like to have it.) For simplicity, we can treat records as NULLABLE_BYTES or COMPACT_NULLABLE_BYTES (depending on the message version).

Tagged fields

Tagged fields are an extension to the Kafka protocol which allows optional data to be attached to messages. The idea is twofold:

If the client of service doesn't understand the tagged field, it'll save it as unknown and ignore it.
If a field is rarely used, its default value can be skipped from transferring.

Have a look, for instance, at this field. It has taggedVersions, which says since which version this field is tagged (in most cases, it's the same version when the field was added).

A tagged field consists of:

The tag of the UNSIGNED_VARINT type.
The data of the COMPACT_BYTES type.

You can find more details about tagged fields in KIP-482.

Let's implement:

@dataclass
class RawTaggedField:
    tag: int
    data: bytes

    @classmethod
    def read(cls, buffer: BinaryIO) -> RawTaggedField:
        tag = read_unsigned_varint(buffer)
        size = read_unsigned_varint(buffer)
        data = read_exact(buffer, size)
        return RawTaggedField(tag=tag, data=data)

    def write(self, buffer: BinaryIO) -> None:
        write_unsigned_varint(self.tag, buffer)
        write_unsigned_varint(len(self.data), buffer)
        buffer.write(self.data)


def read_unknown_tagged_fields(buffer: BinaryIO) -> list[RawTaggedField]:
    # As the tagged field array cannot be null,
    # we need to compensate the length shift that
    # `read_array_length` applies in the compact mode.
    size = read_array_length(buffer, True) + 1
    result = []
    for _ in range(size):
        result.append(RawTaggedField.read(buffer))
    return result


def write_unknown_tagged_fields(
    unknown_tagged_fields: list[RawTaggedField], buffer: BinaryIO
) -> None:
    # As the tagged field array cannot be null,
    # we need to compensate the length shift that
    # `write_array_length` applies in the compact mode.
    write_array_length(len(unknown_tagged_fields) - 1, buffer, True)
    for tf in unknown_tagged_fields:
        tf.write(buffer)

Here they are titled "unknown". Known fields need to be made so inside their structures.

Message structure

The high-level message structure is very straightforward. According to the specification:

RequestOrResponse => Size (RequestMessage | ResponseMessage)
  Size => int32

That is, it's a message itself preceded by its size in bytes. Both request and response messages consist of the header immediately followed by the body. For some reason, this isn't explicitly documented⁴, but you can trust me 🙂 or check the code.

Request and response header

The request header exists in three versions: 0, 1, and 2. They are specified in the protocol as:

Request Header v0 => request_api_key request_api_version correlation_id 
  request_api_key => INT16
  request_api_version => INT16
  correlation_id => INT32

Request Header v1 => request_api_key request_api_version correlation_id client_id 
  request_api_key => INT16
  request_api_version => INT16
  correlation_id => INT32
  client_id => NULLABLE_STRING

Request Header v2 => request_api_key request_api_version correlation_id client_id TAG_BUFFER 
  request_api_key => INT16
  request_api_version => INT16
  correlation_id => INT32
  client_id => NULLABLE_STRING

TAG_BUFFER is the tagged fields mentioned earlier.

Let's implement them as Python data classes:

@dataclass
class RequestHeaderV0:
    request_api_key: int
    request_api_version: int
    correlation_id: int

    def write(self, buffer: BinaryIO) -> None:
        write_int16(self.request_api_key, buffer)
        write_int16(self.request_api_version, buffer)
        write_int32(self.correlation_id, buffer)


@dataclass
class RequestHeaderV1:
    request_api_key: int
    request_api_version: int
    correlation_id: int
    client_id: str

    def write(self, buffer: BinaryIO) -> None:
        write_int16(self.request_api_key, buffer)
        write_int16(self.request_api_version, buffer)
        write_int32(self.correlation_id, buffer)
        write_nullable_string(self.client_id, buffer, False)


@dataclass
class RequestHeaderV2:
    request_api_key: int
    request_api_version: int
    correlation_id: int
    client_id: str
    _unknownTaggedFields: list[RawTaggedField]

    def write(self, buffer: BinaryIO) -> None:
        write_int16(self.request_api_key, buffer)
        write_int16(self.request_api_version, buffer)
        write_int32(self.correlation_id, buffer)
        write_nullable_string(self.client_id, buffer, False)
        write_unknown_tagged_fields(self._unknownTaggedFields, buffer)

As you can see, there are some tagged fields in version 2, there are no expected known fields. If some tagged field is sent erroneously to the broker, it will be ignored.

The response header exists in two versions: 0 and 1. They are specified in the protocol as:

Response Header v0 => correlation_id 
  correlation_id => INT32

Response Header v1 => correlation_id TAG_BUFFER 
  correlation_id => INT32

Let's also implement them:

@dataclass
class ResponseHeaderV0:
    correlation_id: int

    @classmethod
    def read(cls, buffer: BinaryIO) -> ResponseHeaderV0:
        return ResponseHeaderV0(correlation_id=read_int32(buffer))


@dataclass
class ResponseHeaderV1:
    correlation_id: int
    _unknownTaggedFields: list[RawTaggedField]

    @classmethod
    def read(cls, buffer: BinaryIO) -> ResponseHeaderV1:
        return ResponseHeaderV1(
            correlation_id=read_int32(buffer),
            _unknownTaggedFields=read_unknown_tagged_fields(buffer),
        )

We don't implement read for the request headers and write for the response ones. This is for brevity: we're not going to send response headers and receive the request ones in our examples as we're not programming the server side. However, if you're interested in the server side as well, you need to implement both functions (which should be straightforward).

Correlation ID

Note particularly the correlation_id field in the request and response headers. The protocol supports pipelining: the client can have more than one outstanding request per connection. The correlation ID allows it to match responses to requests.

Header version selection

Which version must be used is a function of the API key and message version. It's not currently documented in the protocol guide explicitly⁵.
Use the requestHeaderVersion and responseHeaderVersion functions in the generated class ApiMessageType as the reference.

Sending requests and receiving responses

Now, having all this knowledge and code, let's finally send an ApiVersions request and receive and read a response. ApiVersions is normally the first request that the client sends. It's purpose is to find the API versions and features supported by the broker. We implement the latest version 3.

In the protocol specification, it's defined as:

ApiVersions Request (Version: 3) => client_software_name client_software_version TAG_BUFFER 
  client_software_name => COMPACT_STRING
  client_software_version => COMPACT_STRING

Let's make the data class:

@dataclass
class ApiVersionsRequestV3:
    client_software_name: str
    client_software_version: str
    _unknownTaggedFields: list[RawTaggedField]

    def write(self, buffer: BinaryIO) -> None:
        write_string(self.client_software_name, buffer, True)
        write_string(self.client_software_version, buffer, True)
        write_unknown_tagged_fields(self._unknownTaggedFields, buffer)

And the response:

ApiVersions Response (Version: 3) => error_code [api_keys] throttle_time_ms TAG_BUFFER 
  error_code => INT16
  api_keys => api_key min_version max_version TAG_BUFFER 
    api_key => INT16
    min_version => INT16
    max_version => INT16
  throttle_time_ms => INT32

[api_keys] means "an array of api_keys", where api_keys is the structure defined two lines below.

Converting this to Python data classes:

@dataclass
class ApiVersionsResponseApiKeyV3:
    api_key: int
    min_version: int
    max_version: int
    _unknownTaggedFields: list[RawTaggedField]

    @classmethod
    def read(cls, buffer: BinaryIO) -> ApiVersionsResponseApiKeyV3:
        return ApiVersionsResponseApiKeyV3(
            api_key=read_int16(buffer),
            min_version=read_int16(buffer),
            max_version=read_int16(buffer),
            _unknownTaggedFields=read_unknown_tagged_fields(buffer),
        )


@dataclass
class ApiVersionsResponseV3:
    error_code: int
    api_keys: list[ApiVersionsResponseApiKeyV3]
    throttle_time_ms: int
    _unknownTaggedFields: list[RawTaggedField]

    @classmethod
    def read(cls, buffer: BinaryIO) -> ApiVersionsResponseV3:
        return ApiVersionsResponseV3(
            error_code=read_int16(buffer),
            api_keys=read_array(ApiVersionsResponseApiKeyV3.read, buffer, True),
            throttle_time_ms=read_int32(buffer),
            _unknownTaggedFields=read_unknown_tagged_fields(buffer),
        )

When we speak about arrays, we need to know whether we need compact or non-compact ones. To find this out, let's have a look at the schema definition in ApiVersionsRequest.json. You can see "flexibleVersions": "3+", which means that compact arrays are used starting from version 3 (more on this in README.md in the schema directory). Since we're working with version 3 here, we use compact arrays.

Having the request and response classes implemented, we can send and receive these requests. For this ApiVersions v3 we need the v2 request header and the v0 response header (check the generated ApiMessageType.java). The API key (18) you can find in ApiVersionsRequest.json or in the protocol specification.

def send_request(request_correlation_id: int, sock: socket.socket) -> None:
    buffer = BytesIO()

    # message_size
    # will be filled later
    write_int32(0, buffer)

    header = RequestHeaderV2(
        request_api_key=18,
        request_api_version=3,
        correlation_id=request_correlation_id,
        client_id="test-client",
        _unknownTaggedFields=[],
    )
    header.write(buffer)

    message = ApiVersionsRequestV3(
        client_software_name="test-client",
        client_software_version="1",
        _unknownTaggedFields=[],
    )
    message.write(buffer)

    # Now we know the message size.
    # Return to the beginning of the buffer and put it there.
    request_message_size = buffer.tell() - 4
    buffer.seek(0)
    write_int32(request_message_size, buffer)

    sock.send(buffer.getvalue())


def receive_response(request_correlation_id: int, sock: socket.socket) -> None:
    # The first 4 bytes is the length.
    message_size = read_int32(BytesIO(sock.recv(4)))
    buffer = BytesIO(sock.recv(message_size))

    header = ResponseHeaderV0.read(buffer)
    if header.correlation_id != request_correlation_id:
        raise ValueError()
    pprint(header)

    message = ApiVersionsResponseV3.read(buffer)
    pprint(message)


def main() -> None:
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect(("127.0.0.1", 9092))

    request_correlation_id = 123
    send_request(request_correlation_id, sock)

    receive_response(request_correlation_id, sock)

    sock.close()


if __name__ == "__main__":
    main()

If you run this code, you will see the response header and message printed in the console. Congratulations, you've performed a correct network exchange with the Kafka broker!

You will notice three tagged fields put in _unknownTaggedFields. The read and write methods of the generated ApiVersionsResponseData class and also the message definition in ApiVersionsResponse.json will help you to interpret them. Consider this homework 🙂

In my day job, we developed an open source library Kio. It allows us to do arbitrary Kafka API calls from Python easily. The serialization /deserialization code, like in Kafka itself, is generated from the JSON protocol definitions. The generated code is rigorously tested, including property testing against real Java Kafka code. ↩
Or "message" if you like: some of the schemas are not for API, but e.g. for on-disk data. ↩
The read_exact function has a drawback that it duplicates the data when the underlying buffer is already in memory. However, it's more convenient for education purposes. ↩
I made a PR to fix this. ↩
Again, I made a PR to fix this. ↩

DEV Community: Ivan Yurchenko

Embedding JVM in Rust

Proptest: property testing in Rust

Introduction

Property testing

Proptest

Step by step towards property tests

Step 1. Simplified generated code

Step 2. Property test for serialization/deserialization

Step 3. Deriving Arbitrary

Step 4. Custom strategy for Uuid

Step 5. First run and first failure

Step 6. Custom strategy for Vec<RawTaggedField>

Conclusion

Foot note: testing against real Java Kafka code

Kafka protocol practical guide

Intro

Protocol overview

First steps

Learn Kafka protocol code

Run Kafka in Docker

Inspect protocol with Wireshark

Reading and writing primitive values

Fixed length integer values: INT8, INT16, INT32, INT64, and UINT16

BOOLEAN

FLOAT64

UNSIGNED_VARINT: Variable length integer values

UUID

Strings

Byte arrays

Other arrays

RECORDS

Tagged fields

Message structure

Request and response header

Correlation ID

Header version selection

Sending requests and receiving responses

Step 3. Deriving `Arbitrary`

Step 4. Custom strategy for `Uuid`

Step 6. Custom strategy for `Vec<RawTaggedField>`

Fixed length integer values: `INT8`, `INT16`, `INT32`, `INT64`, and `UINT16`

`BOOLEAN`

`FLOAT64`

`UNSIGNED_VARINT`: Variable length integer values

`UUID`

`RECORDS`