Status

Current state: Voting

Discussion thread: Link

Voting thread: Link

JIRA: KAFKA-20387 - Getting issue details... STATUS

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

RecordHeader stores its value as byte[]. This is the right wire representation, but it makes the API challenging for the types everyone mostly puts in headers (string, int, boolean...):

Construction is tedious:

// String
record.headers().add(new RecordHeader("traceId", traceId.getBytes(UTF_8)));

// Integer
record.headers().add(new RecordHeader("retryCount", ByteBuffer.allocate(4).putInt(3).array()));

// Boolean
record.headers().add("isRetry",
    new byte[] { (byte) (isRetry ? 1 : 0) });

Reading is tedious:

String traceId = new String(header.value(), UTF_8);
int retryCount = ByteBuffer.wrap(header.value()).getInt();
boolean isRetry = header.value()[0] != 0;

Keys and values on ProducerRecord have configurable serializers. Headers have nothing -- every team writes their own wrappers (LoggableRecordHeader, utility methods, byte constants) to make headers usable.

Public Interfaces

Add static factories for construction and typed accessors for reading, covering the common types used in headers. Internally still byte[] with no wire format changes.


Header.java
public interface Header {

    /**
     * Returns the key of the header.
     *
     * @return the header's key; must not be null.
     */
    String key();

    /**
     * Returns the value of the header.
     *
     * @return the header's value; may be null.
     */
    byte[] value();

	// Read accessors
  /**
   * Returns the value of the header decoded as a UTF-8 string.
   *
   * @return the header's value decoded using {@link java.nio.charset.StandardCharsets#UTF_8};
   *         returns {@code null} if the underlying value is {@code null}.
   */
  String stringValue();

  /**
   * Returns the value of the header decoded as a string using the given charset.
   *
   * @param charset the charset to use for decoding; must not be null.
   * @return the header's value decoded using the given charset;
   *         returns {@code null} if the underlying value is {@code null}.
   * @throws NullPointerException if {@code charset} is null.
   */
  String stringValue(Charset charset);

  /**
   * Returns the value of the header interpreted as an 8-bit integer.
   *
   * @return the single byte stored in the header value.
   * @throws IllegalStateException if the underlying value is null or its length is not exactly 1 byte.
   */
  byte byteValue();

  /**
   * Returns the value of the header interpreted as a 16-bit big-endian integer.
   *
   * @return the {@code short} decoded from the header value.
   * @throws IllegalStateException if the underlying value is null or its length is not exactly 2 bytes.
   */
  short shortValue();

  /**
   * Returns the value of the header interpreted as a 32-bit big-endian integer.
   *
   * @return the {@code int} decoded from the header value.
   * @throws IllegalStateException if the underlying value is null or its length is not exactly 4 bytes.
   */
  int intValue();

  /**
   * Returns the value of the header interpreted as a 64-bit big-endian integer.
   *
   * @return the {@code long} decoded from the header value.
   * @throws IllegalStateException if the underlying value is null or its length is not exactly 8 bytes.
   */
  long longValue();

  /**
   * Returns the value of the header interpreted as a single-precision big-endian float.
   *
   * @return the {@code float} decoded from the header value.
   * @throws IllegalStateException if the underlying value is null or its length is not exactly 4 bytes.
   */
  float floatValue();

  /**
   * Returns the value of the header interpreted as a double-precision big-endian float.
   *
   * @return the {@code double} decoded from the header value.
   * @throws IllegalStateException if the underlying value is null or its length is not exactly 8 bytes.
   */
  double doubleValue();

  /**
   * Returns the value of the header interpreted as a boolean.
   * The value must be a single byte: {@code 0x01} for {@code true} or {@code 0x00} for {@code false}.
   *
   * @return the {@code boolean} decoded from the header value.
   * @throws IllegalStateException if the underlying value is null, its length is not exactly 1 byte,
   *         or the byte is not {@code 0x00} or {@code 0x01}.
   */
  boolean booleanValue();
}


Headers.java
public interface Headers extends Iterable<Header> {

  /**
   * Creates and adds a header whose value is the UTF-8 encoding of the given string.
   *
   * @param key of the header to be added; must not be null.
   * @param value of the header to be added; may be null, in which case a header with a null value is added.
   * @return this instance of the Headers, once the header is added.
   * @throws IllegalStateException is thrown if headers are in a read-only state.
   */
  Headers add(String key, String value) throws IllegalStateException;

  /**
   * Creates and adds a header whose value is the encoding of the given string in the given charset.
   *
   * @param key of the header to be added; must not be null.
   * @param value of the header to be added; may be null, in which case a header with a null value is added.
   * @param charset the charset to use for encoding; must not be null.
   * @return this instance of the Headers, once the header is added.
   * @throws NullPointerException  if {@code charset} is null.
   * @throws IllegalStateException is thrown if headers are in a read-only state.
   */
  Headers add(String key, String value, Charset charset) throws IllegalStateException;

  /**
   * Creates and adds a header whose value is the single byte {@code value}.
   *
   * @param key of the header to be added; must not be null.
   * @param value of the header to be added.
   * @return this instance of the Headers, once the header is added.
   * @throws IllegalStateException is thrown if headers are in a read-only state.
   */
  Headers add(String key, byte value) throws IllegalStateException;

  /**
   * Creates and adds a header whose value is the 2-byte big-endian encoding of {@code value}.
   *
   * @param key of the header to be added; must not be null.
   * @param value of the header to be added.
   * @return this instance of the Headers, once the header is added.
   * @throws IllegalStateException is thrown if headers are in a read-only state.
   */
  Headers add(String key, short value) throws IllegalStateException;

  /**
   * Creates and adds a header whose value is the 4-byte big-endian encoding of {@code value}.
   *
   * @param key of the header to be added; must not be null.
   * @param value of the header to be added.
   * @return this instance of the Headers, once the header is added.
   * @throws IllegalStateException is thrown if headers are in a read-only state.
   */
  Headers add(String key, int value) throws IllegalStateException;

  /**
   * Creates and adds a header whose value is the 8-byte big-endian encoding of {@code value}.
   *
   * @param key of the header to be added; must not be null.
   * @param value of the header to be added.
   * @return this instance of the Headers, once the header is added.
   * @throws IllegalStateException is thrown if headers are in a read-only state.
   */
  Headers add(String key, long value) throws IllegalStateException;

  /**
   * Creates and adds a header whose value is the 4-byte big-endian float encoding of {@code value}.
   *
   * @param key of the header to be added; must not be null.
   * @param value of the header to be added.
   * @return this instance of the Headers, once the header is added.
   * @throws IllegalStateException is thrown if headers are in a read-only state.
   */
  Headers add(String key, float value) throws IllegalStateException;

  /**
   * Creates and adds a header whose value is the 8-byte big-endian encoding of {@code value}.
   *
   * @param key of the header to be added; must not be null.
   * @param value of the header to be added.
   * @return this instance of the Headers, once the header is added.
   * @throws IllegalStateException is thrown if headers are in a read-only state.
   */
  Headers add(String key, double value) throws IllegalStateException;

  /**
   * Creates and adds a header whose value is a single byte: {@code 0x01} for {@code true},
   * {@code 0x00} for {@code false}.
   *
   * @param key of the header to be added; must not be null.
   * @param value of the header to be added.
   * @return this instance of the Headers, once the header is added.
   * @throws IllegalStateException is thrown if headers are in a read-only state.
   */
  Headers add(String key, boolean value) throws IllegalStateException;
 
  // rest of the methods
}

Proposed Changes

While RecordHeader is defined under the internals/ package, the public class access is not preventing users from creating a reference via the RecordHeader class. To properly support types, we add new overloaded methods on Header and Headers.

Each ofXXX method serializes the value to byte[] internally using the same encoding as the corresponding Kafka Serializer (UTF-8 for strings, big-endian for numerics, single byte 0x01/0x00 for boolean).

RecordHeader
public static RecordHeader ofString(String key, String value) { ... }
public static RecordHeader ofString(Charset charset) { ... }
public static RecordHeader ofInt(String key, int value) { ... }
public static RecordHeader ofLong(String key, long value) { ... }
public static RecordHeader ofByte(String key, byte value) { ... }
public static RecordHeader ofShort(String key, short value) { ... }
public static RecordHeader ofFloat(String key, float value) { ... }
public static RecordHeader ofDouble(String key, double value) { ... }
public static RecordHeader ofBoolean(String key, boolean value) { ... }
public static RecordHeader ofBytes(String key, byte[] value) { ... }

Typed Accessors

public String stringValue() { ... }
public String stringValue(Charset charset) { ... }
public int intValue() { ... }
public long longValue() { ... }
public byte byteValue() { ... }
public short shortValue() { ... }
public float floatValue() { ... }
public double doubleValue() { ... }
public boolean booleanValue() { ... }

Same pattern: instantiate, deserialize, discard.

The existing byte[] value() method is unchanged.

Null handling: ofString(key, null, charset) (ofStringUtf8) and ofBytes(key, null) create a header with a null byte[] value, consistent with the existing RecordHeader(String, byte[]) constructor which accepts null. Primitive factories (ofInt, ofLong, etc.) cannot accept null since they take primitives. 

Each deserializes the internal byte[]. Throws IllegalStateException if the byte array is the wrong size for the requested type (e.g. calling intValue() on a 7-byte array). Calling a typed accessor on a header with a null value throws IllegalStateException. stringValue() on a null value returns null.

Compatibility, Deprecation, and Migration Plan

Fully backward compatible. All changes are additive with new static factories and new accessor methods. Existing RecordHeader(String, byte[]) constructor and byte[] value() are unchanged. 

Test Plan

  • Unit tests for each static factory (ofString, ofInt, ofLong, ofShort, ofFloat, ofDouble, ofBoolean, ofBytes)
  • Unit tests for each typed accessor, including wrong-size byte array errors
  • Unit tests for wrong-type reading (e.g. create with ofInt, read with stringValue())
  • Unit tests for null values - ofString(key, null), ofBytes(key, null), and typed accessor on null value
  • Unit tests for boundary values (Integer.MAX_VALUE, Integer.MIN_VALUE, Long.MAX_VALUE, Double.NaN, Double.POSITIVE_INFINITY, etc.)
  • Unit tests for empty string (ofString(key, ""))
  • Unit tests for Unicode strings (multi-byte characters, emojis)
  • Unit tests verifying byte[] value() is unchanged
  • Integration tests verifying headers created with static factories survive produce/consume round-trip
  • Integration tests verifying typed accessors work on headers received from consumer

Rejected Alternatives

Configurable header serde on producer/consumer. A config mapping header keys to serializer classes adds complexity and couples header semantics to client config. The common types (string, int, long, short, float, double, boolean) cover the vast majority of header usage and don't need configuration.

Default methods on the Header/Headers interface. While this will help avoid compilation errors after upgrading to a newer AK version, it is better than silently failing for implementers. If users roll their own version of record headers, they must be aware of the type-aware methods and consciously adapt it to their use cases.

Generic serde parameter stored on RecordHeader instance. new RecordHeader(key, value, serializer) makes RecordHeader serde-aware and stores extra references. The static factory approach keeps RecordHeader as a (String, byte[]) tuple with no new instance state. The generic of() / value() methods accept a class, instantiate it, use it, and discard it.

Moving ReadcordHeader/RecordHeaders as part of the public API. Moving the classes out of the internals will break compatibility and force a major version change. The default implementation since the inception of the classes has been within the internals package. We aim to preserve this design choice.

  • No labels