...

throw new IllegalArgumentException("Unknown class for built-in serializer. Supported types are: " +
    "String, Short, Integer, Long, Float, Double, ByteArray, ByteBuffer, Bytes, UUID, Boolean");

public class BooleanSerializer implements Serializer<Boolean> {
	[...]
    public byte[] serialize(final String topic, final Boolean data) { [...]	}
}

public class BooleanDeserializer implements Deserializer<Boolean> {
	[...]
    public Boolean deserialize(final String topic, final byte[] data) { [...] }
}

Proposed Changes

1. Add Boolean Serde to a org/apache/kafka/common/serialization package
   1. Add static public final class to the Serdes.java, update serdeFrom method, add Boolean() method that returns BooleanSerde();
   2. Create BooleanSerializer class in BooleanSerializer.java file, based on what is in NullableValueAndTimestampSerde.java
   3. Create BooleanDeserializer class in BooleanDeserializer.java file, based on what is in NullableValueAndTimestampSerde.java
2. Rewrite NullableValueAndTimestampSerde.java file, package: org.apache.kafka.streams.state.internals so that it uses BooleanSerde from common/serialization instead of having own private class.

...