DE Mastery GitHub

Java for data engineers · JVM-04

Serialization

From your world. the on-the-wire/at-rest format of a table or a Kafka topic (Avro/Parquet) vs an opaque pickle

← All exercises

Avro round-trip and the Flink Kryo-fallback check

Your streaming job will put order events on a Kafka topic as Avro. It will also keep them in Flink state. Two things must be true before you deploy. The records must serialize losslessly through Avro. Your event class must NOT fall back to Flink's slow generic Kryo serializer. In this exercise, you implement both checks. You do not need a Flink runtime.

Implement everything in src/main/java/de/mastery/jvm04/Serialization.java (package de.mastery.jvm04). The file provides the Avro SCHEMA (an Order record with id:long, currency:string, amount:double). It also provides BadEvent, a deliberately broken class that the tests use.

  1. byte[] encode(GenericRecord record) — binary-encode record with SCHEMA using Avro (GenericDatumWriter + EncoderFactory.binaryEncoder). Return the bytes.

  2. GenericRecord decode(byte[] bytes) — decode with the same SCHEMA (GenericDatumReader + DecoderFactory.binaryDecoder). decode(encode(r)) must equal r.

  3. A nested public static class OrderEvent that is a valid Flink POJO. It is a public class with a public no-argument constructor. It has the fields long id, String currency and double amount. Each field is either public or exposed through public getters/setters. (An all-args constructor is also fine, but the no-arg constructor is required.)

  4. boolean isFlinkPojo(Class<?> type) — return true if and only if type satisfies Flink's POJO rules:

    • the class is public (and not an interface/abstract),
    • it has a public no-argument constructor,
    • every non-static declared field is either public, or has both a public getter (getX / isX for boolean) and a public setter (setX).

    isFlinkPojo(OrderEvent.class) must be true. isFlinkPojo(BadEvent.class) must be false. (BadEvent has private fields, no accessors, and no no-arg constructor.)

Constraints:

  • Do not change signatures, the schema, or BadEvent. Keep the package.
  • The round-trip must use Avro binary encoding. The test checks the equality round-trip. It also checks that isFlinkPojo classifies the good and bad types correctly.

Starter code

src/main/java/de/mastery/jvm04/Serialization.java
package de.mastery.jvm04;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;

/**
 * Avro round-trip + a reflective Flink POJO (Kryo-fallback) check. Replace the TODOs.
 *
 * SCHEMA, BadEvent and buildOrder(...) are provided and must not be changed.
 */
public class Serialization {

    public static final Schema SCHEMA = new Schema.Parser().parse(
            "{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"de.mastery\"," +
            "\"fields\":[" +
            "{\"name\":\"id\",\"type\":\"long\"}," +
            "{\"name\":\"currency\",\"type\":\"string\"}," +
            "{\"name\":\"amount\",\"type\":\"double\"}]}");

    /** Binary-encode a GenericRecord with SCHEMA. */
    public static byte[] encode(GenericRecord record) {
        // TODO: GenericDatumWriter + binaryEncoder
        return new byte[0];
    }

    /** Decode bytes with SCHEMA. decode(encode(r)) must equal r. */
    public static GenericRecord decode(byte[] bytes) {
        // TODO: GenericDatumReader + binaryDecoder
        return null;
    }

    /** A valid Flink POJO carrying the same fields as the Avro Order. */
    public static class OrderEvent {
        // TODO: fields + public no-arg constructor + accessors (make this a valid POJO).
    }

    /** True iff `type` satisfies Flink's POJO rules (public class, public no-arg ctor,
     *  every non-static field public or with public getter+setter). */
    public static boolean isFlinkPojo(Class<?> type) {
        // TODO: implement the reflective check.
        return false;
    }

    /** Provided helper: a deliberately NON-POJO class (private fields, no accessors, no no-arg ctor). */
    public static final class BadEvent {
        private final long id;
        private final String currency;
        public BadEvent(long id, String currency) {
            this.id = id;
            this.currency = currency;
        }
        public long computeId() { return id; }     // not a bean getter for `id`
        @Override public String toString() { return currency; }
    }
}

Hints

Hint 1

Round-trip: GenericDatumWriter + binaryEncoder to a ByteArrayOutputStream, then GenericDatumReader + binaryDecoder with the SAME schema; the decoded record must equal the original.

Hint 2

Make OrderEvent a valid Flink POJO: public class, a public no-argument constructor, and either public fields or public getters/setters for every field.

Hint 3

isFlinkPojo via reflection: Modifier.isPublic on the class, a public no-arg constructor exists, and for each non-static field it is public OR both a public getX/isX getter and a public setX setter exist.

Run it locally
uv run de start JVM-04/01-avro-roundtrip
# edit workspace/JVM-04/01-avro-roundtrip/
uv run de test