MASARYK UNIVERSITY
FACULTY OF INFORMATICS
Improving Milena, a Kafka
client for Haskell
BACHELOR'S THESIS
Pavel Kučera
Brno, Spring 2017
Declaration
Hereby I declare that this paper is m y original authorial work, which
I have worked out o n m y o w n . A l l sources, references, a n d literature
used or excerpted during elaboration of this work are properly cited
and listed i n complete reference to the due source.
Pavel Kučera
Advisor: M g r . M a r t i n U k r o p
i
Acknowledgement
A huge thanks goes to m y advisor, Martin Ukrop, for taking me under
his wings, providing valuable knowledge and guidance. Another huge
thanks goes to Peter Trško and Matěj Kollár for giving me advice
on Haskell whenever I got lost i n the u n k n o w n w o r l d of functional
programming.
A s dorky as it is: So long, and thanks for all the fish!
iii
Abstract
This thesis describes extending functionality of M i l e n a , a Haskell
Apache Kafka client library.
Newly supported features are compression, consumer groups and
the Kafka Protocol of version 0.10.1. The implementation makes use of
Haskell language extensions and it complies w i t h the Kafka Protocol,
w h i c h was verified by testing the new functionality against Apache
Kafka reference client.
The newly added functionality was published on G i t H u b and pull
requests were open to make the functionality available i n the m a i n
Milena repository and o n Hackage.
iv
Keywords
Haskell, Apache Kafka, Milena, lenses
v
Contents
1 Introduction 1
2 Apache Kafka 3
2.1 Architecture 4
2.2 Usage as a Messaging System 5
2.3 The Kafka Protocol 5
3 Milena 9
3.1 Type Structure 9
3.2 Serialisation & Deserialisation 13
3.3 Sending a Request to Kafka 14
3.4 Language Extensions 16
4 Lenses 19
4.1 Definition 19
4.2 Package lens 21
4.3 Lenses in Milena 23
5 Supporting Compression 25
5.1 Compressed Messages in the Kafka Protocol 25
5.2 Implementation 26
6 Supporting Multiple Versions of Requests 33
6.1 Phantom Types 34
6.2 GADTs 35
6.3 Creating Generic Request Types 36
6.4 Support of New of Protocol Constructs 38
6.5 New Protocol Versions 40
6.6 Tests 42
7 Supporting Groups 43
7.1 Group Membership Management Process 43
7.2 Implementation 44
7.3 Tests 44
8 Conclusion 47
vii
A Source Code
Bibliography
viii
1 Introduction
This thesis is a result of cooperation w i t h I X P E R T A s.r.o. w h i c h decided
to split its systems written i n Haskell into several micro services
communicating w i t h each other through some messaging system. A f ter
experimenting w i t h other tools, Apache K a f k a 1
was chosen to be
the messaging system. One of the advantages of this decision was the
existence of M i l e n a 2
, an Apache Kafka client written i n pure Haskell.
However, the client library lacked some of the features offered b y
Apache Kafka a n d it d i d not support the newest version of the used
protocol which provided the motivation for this thesis. The implementation
part of this thesis was consulted w i t h Peter Trsko, one of the
IXPERTA's programmers.
The first goal of the thesis is to describe the concept of lenses, which
are used throughout Milena's codebase. The second goal is to extend
the functionality of Milena to support compression, consumer groups
and communication protocol supported by Apache Kafka 0.10.1 and I
should also try to propagate the new functionality into Milena itself
by opening p u l l requests to the official repository hosted o n G i t H u b .
Acceptance of the p u l l requests b y the maintainer w o u l d make the
newly added functionality available o n Hackage.
Apart from the Introduction and Conclusion chapters, this thesis
comprises six other chapters. Chapter 2 briefly presents Apache Kafka,
its architecture a n d h o w it can serve as a messaging system. The
chapter then describes the K a f k a Protocol, w h i c h is crucial to the
development of this thesis as the developed code must communicate
w i t h Apache K a f k a over this protocol. Chapter 3 introduces M i l e n a
itself, its structure, how it handles serialisation and deserialisation of
data and what non-standard Haskell features it uses. Chapter 4 gives
a description of the concept of lenses and how they are used i n Milena.
Chapter 5 focuses on how Apache Kafka handles compression and the
implementation of its support. Chapter 6 describes the changes made
to support the newest version of the Kafka Protocol and what concepts
and language extensions are used to do so. Chapter 7 presents the
modifications adapted to support consumer groups.
1. https://kafka.apache.org
2. https://github.com/adamflott/milena
1
2 Apache Kafka
Apache K a f k a is a distributed streaming platform originally open
sourced i n 2011 a n d currently maintained a n d developed b y the
Apache Software Foundation. The tool is used by several large projects
such as Twitter, Linkedin, Netflix, Spotify and many others [1], which
suggests h i g h level of reliability a n d scalability. To understand what
Apache Kafka can do, we should look at its three key capabilities, as
listed i n Official Apache Kafka Documentation [2]:
1. It lets you publish and subscribe to streams of records. In this respect
it is similar to a message queue or enterprise messaging system.
2. It lets you store streams of records in a fault-tolerant way.
3. It lets you process streams of records as they occur.
Put i n other words, we can send a record2
into Apache Kafka, w h i c h
stores the record persistently, a n d later w e can fetch the very same
record from Apache Kafka. If we send more records to Apache Kafka,
they can be fetched i n the order i n w h i c h they were accepted. The
records can be sent and fetched f r o m entirely different processes or
even machines. A process, w h i c h sends records into Apache Kafka, is
called a Producer a n d a process reading records f r o m Apache K a f k a
is called a Consumer. A simple schema of the data flow is depicted i n
fig. 2.1.
The provided capabilities create two broad classes of application
for Apache Kafka [2]:
1. Building real-time streaming data pipelines that reliably get data between
systems or applications3
.
2. Building real-time streaming applications that transform or react to
the streams of data4
.
1. https://kafka.apache.org/
2. A record can be anything from a single byte to a whole binary file.
3. For example, several services could push their logs into Apache Kafka and a
different application could read the logs from Apache Kafka and visualise them.
4. In a system processing photos, one service could receive a photo and send a
notification about the new photo into Apache Kafka. A different service could read
these notifications and crop the pictures according to the system's needs.
3
2. A P A C H E K A F K A
Producer #1 Producer #2
Kafka Cluster
Consumer # 1 Consumer #2 Consumer #3
Figure 2.1: Data flow i n Apache Kafka
To understand h o w K a f k a works, w e also need to look at a few concepts
[2]:
1. Kafka is run as a cluster on one or more servers.
2. The Kafka cluster stores streams of records in categories called topics.
3. Each record consists of a key, a value, and a timestamp.
The first concept makes Apache Kafka a distributed system. The second
concept gives us the option not to mix data, which do not belong
together, and it also allows Apache Kafka to perform better, as different
topics can be handled b y different processes. The third concept
discloses what a record is - a value marked by a key, for which Apache
Kafka remembers the time of creation.
2.1 Architecture
To provide the promised functionality, Kafka has different core APIs,
out of w h i c h w e need to concentrate o n the Producer API a n d Consumer
API, w h i c h allow publishing a stream of records to K a f k a and
allow reading a stream of records f r o m K a f k a respectively5
[2]. The
information i n this section is based o n [2].
Whenever a producer publishes a record or a consumer consumes
a record, they always w o r k w i t h a particular topic, w h i c h is an abstraction
over the streams of records and it provides categorisation
5. The others APIs are the Streams API and the Connector API, but they do not
concern the topic of the thesis and thus are not described.
4
2. A P A C H E K A F K A
of data. Topics are further divided into partitions and a producer has
to choose the partition it wants to publish a record into. Similarly, a
consumer must consume a record from a particular partition. Reads
and writes from a n d to different partitions can be done i n parallel,
effectively making partitions the unit of parallelism.
To further support parallel processing, Apache Kafka introduces a
concept of consumer groups. A consumer group consists of several
consumers a n d K a f k a ensures that a record is delivered only to one
consumer per group - if all consumers are a part of the same group,
record processing w i l l be spread across all of the consumers a n d if
every consumer is i n a different group, every record w i l l be processed
by every single one of the consumers.
2.2 Usage as a Messaging System
There are many uses for Apache Kafka, for example messaging system,
log aggregation, stream processing, event sourcing or commit log [2],
but this thesis concentrates o n using K a f k a as a messaging system.
That makes a node of a Kafka cluster a message broker and individual
records are messages. Kafka combines the two traditional messaging
models - queuing and publish-subscribe [2].
W h e n a message is sent to Apache Kafka using the Producer A P I ,
it is assigned a unique number called offset6
. W h e n messages are
fetched from Apache Kafka using the Consumer A P I , they are always
fetched from a specific starting offset7
.
2.3 The Kafka Protocol
To communicate w i t h its clients, Kafka uses a custom binary protocol
over TCP, which is described i n Kafka Protocol Guide [3] and A Guide To
The Kafka Protocol [4]. A l l information about the protocol is based o n
these two sources. The protocol is described using Backus-Naur form,
shortly B N F , copying the original definition from the named sources.
6. Offset is unique in terms of a specific partition of a specific topic
7. A consumer must remember its position/offset if it should avoid re-reading
messages.
5
2. A P A C H E K A F K A
The protocol defines its APIs as request/response message pairs.
Every message is delimited b y its size and is made of the following
primitive types.
Note on Naming
Kafka Protocol Guide [3] is somewhat chaotic i n naming conventions sometimes
names have their first letter capitalised, sometimes they are
written i n capitals and sometimes they are written i n lower case. This
thesis keeps uniform naming for constructs from the Kafka Protocol
and thus the naming may differ i n capitalisation.
Fixed Width Primitives
The fixed w i t h primitives used i n the Kafka Protocol are INT8, INT 16,
INT32 and INT64, w h i c h are signed integers w i t h the given bit precision.
The integers are stored i n big endian order.
Variable Length Primitives
The variable length primitives used i n the Kafka Protocol are BYTES
and STRING, both of them comprise two parts. First their length N
is stated b y a signed integer and then N bytes of content follows.
Type BYTES uses an INT32 to state its length, STRING uses an INT 16. A
length of -1 indicates NULL. W h e n a STRING is nullable, it is referenced
as NULLABLE_STRING i n the protocol definition.
Arrays
Arrays are used to w r a p repeated structures and one array always
contains only one k i n d of a structure. A n array consists of two parts.
First the number N of its elements is stated i n an INT32 value, then
N repetitions of the structure follows8
. The structure repeated i n the
array can be made u p of other types.
8. There is one exception to this rule, an array of MessageSetMembers in the
MessageSet structure is not prefixed by its length.
6
2. A P A C H E K A F K A
Complex Types
Complex, nested types can be defined by combining the already mentioned
types. For example, an imaginary Foo type c o u l d be defined
as:
Foo => count t e x t [tag]
count => INT16
t e x t => STRING
tag => STRING
W h e n a value may be of multiple types, they are separated w i t h " I",
as i n the following example:
Bar => Foo I FooBar
Requests
A l l requests extend the following header:
Request
=> s i z e api_key a p i _ v e r s i o n c o r r e l a t i o n _ i d c l i e n t _ i d
s i z e => INT32
api_key => INT16
a p i _ v e r s i o n => INT16
c o r r e l a t i o n _ i d => INT32
c l i e n t _ i d => NULLABLE_STRING
Where s i z e gives the size of the whole request, api_key defines the
type of the request - for example w h e n a p i k e y equals to zero, it
indicates a Produce Request, w h e n it equals to one, it indicates a
Fetch Request etc. The a p i _ v e r s i o n value indicates the version of
the request. c o r r e l a t i o n _ i d is an integer supplied by the user, which
is passed back w i t h the response and can be used to match a request
with its response. The c l i e n t _ i d value is user-specified identification
of the client.
After the header, the actual request follows and a request can be
defined i n the following manner:
7
2. A P A C H E K A F K A
Heartbeat Request
=> group_id group_generation_id member_id
group_id => STRING
group_generation_id => INT32
member_id => STRING
Responses
A l l responses extend the following header:
Response => s i z e c o r r e l a t i o n _ i d
s i z e => INT32
c o r r e l a t i o n _ i d => INT32
s i z e gives the size of the w h o l e response a n d c o r r e l a t i o n _ i d is
the same c o r r e l a t i o n i d supplied i n the request resulting i n the
response. After headers, a specific response follows. A response might
be defined as, for instance:
Heartbeat Response => error_code
error_code => INT16
Producing and Consuming Messages
To produce a message, a producer must send a Produce Request. To
retrieve a message, a consumer must send a Fetch Request. M e s sages
themselves are represented b y structure Message. However,
they are further w r a p p e d i n MessageSetMembers a n d grouped into
MessageSets. These types are then a part of the Produce a n d Fetch
requests and their respective responses.
8
3 Milena
Milena is a client for Apache Kafka written i n pure Haskell. The library
is separated into four modules1
:
• Network.Kafka
• N e t w o r k . K a f k a . P r o t o c o l
• Network.Kafka.Producer
• Network.Kafka.Consumer
The core of the library is the Protocol module. It defines data types and
functions enabling serialisation of Haskell data type values into byte
strings understood by K a f k a and moreover it enables interpretation
of byte strings generated by Kafka and supports their deserialisation
into values of Haskell data types. The module also defines functions
able to send a request over a specified connection and interpret the
received response.
O n top of the Protocol module, the Kafka module adds functions
enabling us to connect to a specific K a f k a broker. C o m b i n e d , these
two modules provide all the functionality needed for communication
w i t h a K a f k a broker, but the p r o v i d e d A P I is rather low-level. The
remaining modules - Producer and Consumer - add higher level A P I
for producing and consuming messages respectively. The structure of
module dependencies is depicted i n fig. 3.1.
It is worth noting that Milena publicly exports the whole Protocol
module, w h i c h makes almost any change to the module a potential
backwards compatibility break, but also allows anyone to implement
their o w n producers or consumers.
3.1 Type Structure
To understand how Milena works, we first need to look at the Protocol
module and h o w it relates to the K a f k a Protocol. Put shortly, the
1. The modules will be referenced only by the last part of their name further in the
thesis.
9
3- M I L E N A
Kafka. Protocol >- Kafka
Kafka.Producer Kafka.Consumer
Figure 3.1: Milena's module structure. The arrows point from the
imported module to the importing module.
module represents a Haskell definition of the Kafka Protocol and its
type structure very m u c h mimics the B N F protocol definition.
Primitive Types
These types represent the primitive types of the Kafka Protocol. A l l
of the primitive types have their S e r i a l i z a b l e and D e s e r i a l i z a b l e
instances defined2
.
Fixed W i d t h Primitives
To represent the INT8, INT 16, INT32 and INT64 K a f k a Protocol p r i m itives,
M i l e n a uses data types Int8, I n t l 6 , Int32 and Int64 respectively
from standard Date. Int module.
Variable Length Primitives
To represent a bytes value, Milena declares a custom Kaf kaBytes data
type:
newtype KafkaBytes = KBytes {
_kafkaByteString : : B y t e S t r i n g
} d e r i v i n g (Show, E q , I s S t r i n g )
Similarly, a s t r i n g value is represented by a custom Kaf k a S t r i n g data
type:
newtype K a f k a S t r i n g = K S t r i n g {
_ k S t r i n g : : B y t e S t r i n g
} d e r i v i n g (Show, E q , Ord, I s S t r i n g )
2. The (de)serialisation process is thoroughly explained in section 3.2.
10
3- M I L E N A
We should note that both types have their instance of the I s S t r i n g
type class3
and thus any S t r i n g value can be automatically converted
into either Kaf kaBytes or Kaf k a S t r i n g value, based on the expected
type-
Arrays
Arrays are represented by standard lists.
Series of Values
To represent a series of n values, Milena uses n-tuple. For instance, if
a structure was defined i n the Kafka Protocol as:
S t r u c t u r e => INT8 INT16 INT32
then i n M i l e n a the structure w o u l d be represented by a triple of the
following type:
S t r u c t u r e : : ( I n t 8 , I n t l 6 , Int32)
Using Primitive Types
The primitive types, such as Int8, are usually wrapped i n some named
intermediary type to make working w i t h them understandable. For
example, Milena defines the following types:
newtype ApiKey = ApiKey I n t l 6 d e r i v i n g . . .
newtype C l i e n t l d = C l i e n t l d K a f k a S t r i n g d e r i v i n g . . .
These types clearly state the semantics of their value and thanks to
derived type class instances, working w i t h them is as lightweight as if
the primitive types representatives were used directly.
Complex Types
Complex and nested structures are created by combining the intermediary
types. In the following example, we can see a Haskell definition
of a complex structure called Of f setFetchRequest. To illustrate
the connection to the Kafka Protocol definition, the definition of
Of f setFetch Request from [3] follows immediately:
3. The type class is further described in section 3.4.
11
3- M I L E N A
newtype ConsumerGroup = ConsumerGrovrp K a f k a S t r i n g
d e r i v i n g . . .
newtype TopicName = TName {
_tName : : K a f k a S t r i n g
} d e r i v i n g . . .
newtype P a r t i t i o n = P a r t i t i o n Int32 d e r i v i n g . . .
newtype OffsetFetchRequest = OffsetFetchReq (
ConsumerGroup,
[(TopicName, [ P a r t i t i o n ] ) ]
) d e r i v i n g (Show, E q , S e r i a l i z a b l e )
O f f s e t F e t c h Request ( V e r s i o n : 0) => group_id [topics]
group_id => STRING
t o p i c s => t o p i c [ p a r t i t i o n s ]
t o p i c => STRING
p a r t i t i o n s => p a r t i t i o n
p a r t i t i o n => INT32
Even though the naming does not fully correspond, the relation should
be clearly visible.
Requests & Responses
Requests and responses are defined as complex data types. Naming of
the data types corresponds w i t h naming i n the Kafka Protocol, but data
constructor names are shortened. For example, if the Kafka Protocol
defined SomeRequest and SomeResponse, M i l e n a w o u l d define them
as:
newtype SomeRequest = SomeReq . . .
newtype SomeResponse = SomeResp . . .
Message-Related Types
A single message is represented by data type Message. A message i n
a set is represented by a MessageSetMember and a set of messages is
represented by MessageSet. A l l the aforementioned types are used
12
3- M I L E N A
for both producing and consuming messages and can be found in the
related requests and responses.
3.2 Serialisation & Deserialisation
To serialise and deserialise values M i l e n a uses package cereal4
". Two
crucial types provided by cereal are the Get5
and P u t 6
monads and the
functions working w i t h them. The core (de)serialisation functionality
lies i n type classes S e r i a l i z a b l e and D e s e r i a l i z a b l e defined in the
Protocol module.
A value w i t h an instance of the S e r i a l i z a b l e type class is serialised
when a request is being sent and based on the expected response
type, the response is deserialised according to its D e s e r i a l i z a b l e
instance. The (de)serialisation processes rely heavily on nesting - i n stances
of the aforementioned type classes are defined for primitive
types, such as Int8, and the instances for complex types recursively
call the s e r i a l i z e or d e s e r i a l i z e functions when working w i t h other
types. To illustrate the whole instance chain, let us look at the definitions
of the type classes and some of their instances:
import Control.Monad
import D a t a . B i n a r y . G e t
import D a t a . B i n a r y . P u t
import D a t a . I n t
c l a s s S e r i a l i z a b l e a where
s e r i a l i z e : : a -> Put
c l a s s D e s e r i a l i z a b l e a where
d e s e r i a l i z e : : Get a
instance S e r i a l i z a b l e Int8 where
s e r i a l i z e = putWord8 . f r o m l n t e g r a l
instance D e s e r i a l i z a b l e Int8 where
d e s e r i a l i z e = fmap f r o m l n t e g r a l getWord8
4. http://hackage.haskell.org/package/cereal
5. Get is an Exception and State monad [5]
6. Put is a Writer monad [6]
13
3- M I L E N A
instance ( S e r i a l i z a b l e a, S e r i a l i z a b l e b) =>
S e r i a l i z a b l e ((,) a b) where
s e r i a l i z e (x, y) = s e r i a l i z e x >> s e r i a l i z e y
instance ( D e s e r i a l i z a b l e a, D e s e r i a l i z a b l e b) =>
D e s e r i a l i z a b l e ((,) a b) where
d e s e r i a l i z e = l i f t M 2 (,) d e s e r i a l i z e d e s e r i a l i z e
instance S e r i a l i z a b l e a => S e r i a l i z a b l e [a] where
s e r i a l i z e xs = do
l e t 1 = f r o m l n t e g r a l (length xs) : : Int32
s e r i a l i z e 1
mapM_ s e r i a l i z e xs
instance D e s e r i a l i z a b l e a => D e s e r i a l i z a b l e [a] where
d e s e r i a l i z e = do
1 <- d e s e r i a l i z e : : Get Int32
r e p l i c a t e M ( f r o m l n t e g r a l 1) d e s e r i a l i z e
In the instances for a tuple and list we can see the (de)serialisation
functions being called recursively. The whole (de)serialisation process
of a request and response takes place in function doRequest' defined
in the Protocol module.
We should note that the d e s e r i a l i z e function does not take any
arguments, but rather reads input f r o m an underlying State monad,
w h i c h is initialised w i t h its input i n the doRequest' function.
3.3 Sending a Request to Kafka
A l t h o u g h Milena provides a high level A P I for sending requests to a
Kafka broker in its Producer and Consumer modules, it is better to look
at the actual functions which serialise a request, send it and interpret
the response. The core of this functionality is function doRequest'
defined i n the Protocol module. To understand it better, let us see its
type signature:
doRequest' : : ( D e s e r i a l i z a b l e a, MonadIO m) =>
C o r r e l a t i o n l d
-> Handle
-> Request
-> m ( E i t h e r S t r i n g a)
14
3- M I L E N A
We can skip the first parameter . The second parameter is a standard
System. 10. Handle representing a connection to a Kafka broker. In the
third parameter, the function accepts a request, w h i c h is serialised
and written into the given Handle. The response is deserialised into a
data type determined by the a type variable, w r a p p e d i n the E i t h e r
data type8
and returned from the function. Practically, we can provide
any request and expect any response:
expectA : : MonadIO m => m ( E i t h e r S t r i n g SomeResponse)
expectA = doRequest' . . .
expectB : : MonadIO m => m ( E i t h e r S t r i n g AnotherResponse)
expectB = doRequest' . . .
However, we should note that the expected response should match
the provided request, otherwise deserialisation w o u l d most probably
fail during runtime9
.
To create a request accepted b y d o R e q u e s t w e need to create a
value of Request type, but for that we need to utilise another type RequestMessage.
The two types are defined as:
newtype Request = Request
( C o r r e l a t i o n l d , C l i e n t l d , RequestMessage)
d e r i v i n g (Show, Eq)
data RequestMessage =
MetadataRequest MetadataRequest
I OffsetFetchRequest OffsetFetchRequest
d e r i v i n g (Show, Eq)
The RequestMessage type defines a data constructor for every supported
request1 0
and serves as a generalisation of requests. The Request
7. Its meaning is explained in section 2.3.
8. Following common conventions, if an error occurs, the function returns Left
String, rather than Right a.
9. In theory, it is possible that a response of some type A could also be interpreted
as a response of type B, but such behaviour is highly unlikely and relying on it would
go against good programming practises.
10. Metadata Request, Produce Request, Fetch Request, Offset Request, OffsetCommit
Request, OffsetFetch Request, GroupCoordinator Request
15
3- M I L E N A
type then groups together data needed for sending a request and both
data types are necessary to create a request accepted by doRequest'.
The Protocol m o d u l e also defines data type ReqResp a and function
doRequest which encapsulate the aforementioned types and functions
and provide slightly more convenient, but limited, A P I for sending
requests.
3.4 Language Extensions
Milena depends on several language extensions to keep its code shorter
and more readable. The extensions are OverloadedStrings, GADTs, GeneralizedNewtypeDeriving,
TemplateHaskell and RankNTypes, all of which
need to be enabled by, for example, using a standard LANGUAGE pragma
in the library code. This section provides a brief description of every
of the language extensions and w h y they are needed i n Milena.
OverloadedStrings
This extension allows overloading the string literal. Whereas normally,
a string literal has type S t r i n g , w i t h OverloadedStrings the type
changes to [7]:
"" : : ( I s S t r i n g a) => a
The I s S t r i n g type class is defined as [7]:
c l a s s I s S t r i n g a where
fromString : : S t r i n g -> a
U s i n g this type class, value of an arbitrary type can be constructed
by using standard string syntax as long as an instance of I s S t r i n g is
declared for the type.
GADTs
This extension allows data constructors to have richer return types [7].
The capabilities of this extension are more thoroughly described i n
section 6.2.
16
3- M I L E N A
GeneralizedNewtypeDeriving
This extension widens the range of instances w h i c h can be derived
for abstract types declared by newtype [7]. In Milena, this extension is
leveraged to derive type class instances for intermediary data types
w h i c h serve only as wrappers for primitive types.
TemplateHaskell
This extension brings compile-time meta-programming [7] into Haskell
and is used to automatically derive lenses1 1
.
RankNTypes
This extension enables arbitrary-rank explicit universal quantification
i n types [7]. U s i n g this extension allows M i l e n a to define complex
short-cut functions composing lenses.
11. The concept of lenses and their usage in Milena is described in chapter 4.
17
4 Lenses
"The concept of lenses in functional programming offers a way of focusing
on a particular part of a, possibly nested, data structure or container.
This focused part is called the view. The container is called the source." —
Steckermeier [8, p. 1]
Lenses allow both accessing and modifying data structures and i n
Haskell their usage is motivated by deficiencies i n the language itself,
namely by tedious access to nested values. Let us look at an example:
data Message = Message ( I n t , S t r i n g )
getValue : : Message -> S t r i n g
getValue (Message (_, v)) = v
setValue : : Message -> S t r i n g -> Message
setValue (Message (k, _)) v = Message (k, v)
The example defines a type Message and two functions able to access a
value nested inside of the data type. Both functions share very similar,
boilerplate code w h i c h extracts parts of the Message type and w h i c h
would get tediously long if the values were nested even further, e. g. i n
a quadruple wrapped i n another type. The concept of lenses addresses
this issue.
4.1 Definition
In [8], a basic lens is defined as a data type containing two functions:
data Lens s v = Lens (s -> v) (v -> s -> s)
The first function is a getter get, which returns view v, w h e n applied
on source s [8]. The second function is a setter put, which replaces the
v i e w i n the source given to the function i n the second argument by
the v i e w given to the function i n its first argument [8]. get and put
are required to be total functions.
Lenses can be categorised, based o n the rules they follow. The
following rules characterise well-behaved lenses [9, p. 1]:
19
4. LENSES
1. If a v i e w value b is put into a source a by using put, the same
value can be retrieved by using get.
get (put b a) = b
2. If a v i e w value is retrieved f r o m a source and put back, the
container does not change.
put (get a) a = a
Another important category of lenses are very well-behaved lenses,
w h i c h require lenses to be well-behaved and to comply w i t h a third
rule [9, p. 1]:
3. A n application of put does not affect any later applications of
put.
put c (put b a) = put c a
Although these rules are not, strictly speaking, necessary, if they hold
for a lens, it behaves very predictably and thus is easier to use. The
lens itself is just the first step, additionally, two operations - view and
set - extracting the getter and setter are defined [8]:
view : : Lens s v -> (s -> v)
view (Lens get put) = get
set : : Lens s v -> (v -> s -> s)
set (Lens get put) = put
By using the defined lens type, w e can create a lens for virtually any
type from tuples to complex data structures, and by using the defined
operations, w e can easily access data nested i n such structures. For
example, we can define a lens for the Message type from the beginning
of this chapter:
messageValue : : Lens Message S t r i n g
messageValue = Lens get put
where get (Message (_, v)) = v
put v (Message (k, _)) = Message (k, v)
20
4. LENSES
getValue : : Message -> S t r i n g
getValue = view messageValue
setValue : : Message -> S t r i n g -> Message
setValue m v = set messageValue v m
This is the very basic idea of lenses, but i n this basic form, lenses would
not be that useful - a lot of boilerplate code w o u l d still be needed and
such lenses can not be chained using the dot operator. However, the
basic concept can be extended to overcome these issues and there are
many Haskell packages addressing this issue.
It is w o r t h noting that these packages define lenses differently,
they can define more categories of lenses, not just well-behaved a n d
very well-behaved, but they keep the functionality of being able to access
nested data.
4.2 Package lens
lens1
is one of the packages p r o v i d i n g a generalised version of the
concept of lenses. Apart from the basic lens functionality, the package
provides other features such as [10,11]:
• Automatic derivation of lenses2
• Variety of pre-packed lenses for c o m m o n data types
• Lenses composable v i a the dot operator
• Infix operators for c o m m o n operations
To fully understand the package capabilities and its syntax, it is best
to look at [10] a n d the rest of the G i t H u b W i k i of the package. It
is not necessary to k n o w the lens package to understand the code
examples further i n this thesis, but a basic knowledge is needed to
understand Milena's code. The features used by Milena are described
in the following subsections based o n [10,11].
1. http://hackage.haskell.org/package/lens
2. For this feature, the TemplateHaskell language extension has to be enabled.
21
4. LENSES
Automatic Derivation of Lenses
If the TemplateHaskell language extension is enabled, lens c a n automatically
derive lenses for data types w i t h n a m e d fields b y using
function makeLenses. The function automatically creates lenses for
named fields prefixed by an underscore. For example:
{-# LANGUAGE TemplateHaskell #-}
data M a g i c i a n = M a g i c i a n { powers : : [String] }
makeLenses 1
' M a g i c i a n
This w o u l d automatically create a lens powers providing access to the
_powers field of the Magician type.
Pre-packed Lenses
lens come w i t h pre-defined lenses for common Haskell data types like
n-tuples, traversables and foldables. Milena uses mainly the n-tuples
related functions w h i c h provide access to individual members of an
n-tuple i n the following manner:
type T r i p l e = ( B o o l , I n t , S t r i n g )
f i r s t : : T r i p l e -> Bool
f i r s t = view _1
second : : T r i p l e -> I n t
second :
= view _2
t h i r d : : T r i p l e -> S t r i n g
t h i r d = view 3
Composing Lenses
Lenses generated by lens or created according to lens' definition can
be composed using the dot operator [11]:
22
4. LENSES
{-# LANGUAGE TemplateHaskell #-}
data P o i n t = P o i n t { _x : : I n t , _y : : I n t }
data L i n e = L i n e { s t a r t : : P o i n t , end : : P o i n t }
makeLenses 1
' P o i n t
makeLenses ' ' L i n e
l i n e S t a r t X : : L i n e -> Int
l i n e S t a r t X = view ( s t a r t . x)
A n y number of lenses can be composed as long as they form a valid
chain3
.
Infix Operators
To further shorten code using lens, the package defines infix operators
for most of its operations. For view the operator is " ~ . " and for set the
operator is " . ~". Especially when combined with lens composition, the
operators provide a very short syntax for accessing nested data. The
following example expands the example from the previous subsection
and shows the usage of infix operators:
l i n e S t a r t X ' : : L i n e -> Int
l i n e S t a r t X ' 1 = 1 ~ . s t a r t . x
changeX : : P o i n t -> I n t -> P o i n t
changeX p o i n t value = x .~ value point
4.3 Lenses in Milena
Milena's type architecture is designed after the B N F definition of
Kafka Protocol a n d it brings its benefits, like easy (de)serialisation,
but it also means that the type structure is inherently nested - both
requests a n d responses contain primitive types, lists, n-tuples, all
of w h i c h are usually nested i n a complex of other types. To avoid
3. A n invalid chain would be discovered during type check.
23
4. LENSES
constant unpacking and repacking of data constructors, Milena uses
lenses from the lens package.
In the Protocol module, M i l e n a lets lens automatically derive
lenses for types related to Produce and Fetch requests and their respective
responses. Furthermore, it defines several functions composing
the automatically generated lenses to create "short-cut" functions
w h i c h take a structure a n d either extract a small part of it or
return another lens. The derived lenses a n d defined short-cuts are
used throughout the Kafka, Producer a n d Consumer modules. To
illustrate the usage, we can look at the following example:
{-# LANGUAGE TemplateHaskell #-}
newtype Message = Message { _messageFields
: : ( C r c , MagicByte, A t t r i b u t e s , Key, Value)
} d e r i v i n g (Show, E q , D e s e r i a l i z a b l e )
makeLenses 1
'Message
The code defines a data type Message a n d then lets the lens package
derive lenses for Message by calling makeLenses "Message. In this
case, automatic derivation creates one lens represented b y a function
messageFields, which provides access to the quintuple stored i n
Message. If we wanted to access the quintuple using lenses, we could
do so i n the following manner:
quintuple : : Message
-> ( C r c , MagicByte, A t t r i b u t e s , K e y , Value)
quintuple = view messageFields
Short-cut functions are then defined b y composing lenses such as
messageFields:
messageValue : : Lens1
Message Value
messageValue = messageFields . _5
Which allows retrieving a value from a message directly, by using the
composed lens:
value : : Message -> Value
value = view messageValue
24
5 Supporting Compression
Apache K a f k a supports end-to-end block compression allowing a
producer to compress only selected messages. Compressed messages
are then sent to a Kafka broker, w h i c h passes them to the consumers
in the compressed form [12]. Consequently, using compression w i t h
Kafka brings additional C P U load while possibly taking load off from
IO components, e.g. the disks and the network. U s i n g compression
is then beneficial the most w h e n IO components are the bottleneck
and there is unused C P U power. To fully leverage the benefits of
compression, [4] recommends compressing messages i n batches.
A s of spring 2017, Apache K a f k a supports three compression
codecs - Gzip, Snappy and LZ4. This thesis adds support only for Gzip
as neither Snappy, nor L Z 4 have their packages released i n Stackage
LTS 7.19 used by Milena [13]. However, adding support for any other
compression codec should be only a matter of a few lines of code now.
5.1 Compressed Messages in the Kafka Protocol
The information about the representation of compressed messages in
this chapter is based on Separate part of the Apache Kafka documentation,
related to compression [12] and A Guide To The Kafka Protocol [4].
The base unit of compression i n the Kafka protocol is a message set1
.
When compressing any number of messages, they first need to be put
into a regular message set, which is then compressed and represented
as a single message2
. The message containing the compressed message
set is put into a new message set. This message set is then included i n
a Produce Request and sent to Apache Kafka.
W h e n querying a K a f k a Broker for messages, its response may
contain both compressed and uncompressed data. To differentiate
messages containing compressed content, such messages need to set
their headers accordingly. N a m e l y they need to state the used compression
codec. The information about the used compression codec
is stored i n the three lowest bits of the A t t r i b u t e s field of a message.
1. Structure MessageSet of the Kafka Protocol.
2. This effectively makes the definition of a message recursive.
25
5. SUPPORTING C O M P R E S S I O N
Leaving these bits set to 0 implies no compression, setting them to 1
implies G z i p , setting them to 2 implies Snappy and setting them to 3
implies L Z 4 .
5.2 Implementation
The implementation can be split into three consequent steps:
1. Milena must be able to properly read and write the value of the
A t t r i b u t e s field w h e n it contains information about the used
compression codec.
2. M i l e n a must be able to decompress a message containing a
compressed message set and interpret it as separate messages.
3. M i l e n a must be able to compress messages and properly put
them into a wrapping message. These messages must be understood
by clients written i n different programming languages.
Attributes Manipulation
Because Milena d i d not support compression at all, it d i d not need to
operate w i t h the value of the A t t r i b u t e s field and thus it stored the
attributes' value as a whole in an Int8 value. Although this technically
allowed both reading and writing compression information, such
manipulation w o u l d be very cumbersome as it w o u l d always require
using bitwise operations, not allowing to take full advantage of Haskell
capabilities, e.g. pattern matching. Also, using an Int8 w o u l d require
the programmer using M i l e n a to k n o w the K a f k a Protocol i n detail.
Hence the A t t r i b u t e s type h a d change to simplify w o r k i n g w i t h
compression parameters.
A s a first step, a new type was introduced to represent a compression
codec:
data CompressionCodec = NoCompression
I Gzip
d e r i v i n g (Show, Eq)
26
5. SUPPORTING C O M P R E S S I O N
Then, the A t t r i b u t e s type was adjusted to carry compression information.
The initial idea was to m o d i f y it to bear compressing a n d
decompressing functions, w h i c h w o u l d be able to compress a message
along w i t h information about the used codec a n d vice versa.
This w o u l d allow a program using M i l e n a to provide its o w n compression/decompression
functions, but after some consideration, this
solution was rejected for two reasons:
1. Because A t t r i b u t e s are b o u n d to only one message, the solution
w o u l d not allow compressing more than one message at
once, meaning it w o u l d not support the full extent of compression
capabilities i n Apache Kafka. It w o u l d also go against the
recommendation of A Guide To The Kafka Protocol [4], w h i c h is
to batch messages w h e n using compression.
2. It w o u l d not allow keeping precise instances of Show a n d Eq
type classes for A t t r i b u t e s as there is no standard way to show,
nor compare two functions.
The next, and final, idea was to adjust the signature of the A t t r i b u t e s
type to carry a value of CompressionCodec and thus the type signature
changed from:
newtype A t t r i b u t e s = A t t r i b u t e s I n t 8 d e r i v i n g . . .
to:
data A t t r i b u t e s = A t t r i b u t e s {
_compressionCodec : : CompressionCodec
} d e r i v i n g (Show, Eq)
Because of this change, neither the instance of S e r i a l i z a b l e , nor
D e s e r i a l i z a b l e could be derived any more a n d h a d to be implemented
manually. The S e r i a l i z a b l e instance is especially simple as
it only converts the CompressionCodec value into its corresponding
number:
27
5. SUPPORTING C O M P R E S S I O N
instance S e r i a l i z a b l e A t t r i b u t e s where
s e r i a l i z e = s e r i a l i z e . b i t s
where b i t s : : A t t r i b u t e s -> Int8
b i t s = codecValue . _compressionCodec
codecValue : : CompressionCodec -> Int8
codecValue NoCompression = 0
codecValue Gzip = 1
The instance of D e s e r i a l i z a b l e has to deal w i t h potentially unknown
compression codecs, but its core is also i n comparing numbers and
converting them into a value of CompressionCodec:
instance D e s e r i a l i z a b l e A t t r i b u t e s where
d e s e r i a l i z e = do
i <- d e s e r i a l i z e : : Get Int8
codec <- case compressionCodecFromValue i of
Just c -> r e t u r n c
Nothing -> f a i l $
"Unknown compression codec value found i n : "
++ show i
r e t u r n $ A t t r i b u t e s codec
compressionCodecFromValue : : Int8
-> Maybe CompressionCodec
compressionCodecFromValue i I eq 1 = J u s t Gzip
I eq 0 = J u s t NoCompression
I otherwise = Nothing
where eq y = i .&. y == y
Full instances are s h o w n to point out an issue - the compression
codec constants are defined i n two places simultaneously w h i c h is
generally considered a bad practise. Unfortunately, I was not able to
solve this issue satisfactorily myself and even after a consultation w i t h
Peter Trsko, we decided to keep the provided solution for simplicity.
However, it is definitely one of the weak spots of the implementation.
28
5. SUPPORTING C O M P R E S S I O N
Decompressing Messages
Because a message definition is recursive, decompression may take
place only after obtaining a whole message. Its header must be checked
and if it indicates compression, the content of the message can be
decompressed into one or multiple different messages.
There were two possible options of placing the decompressing
mechanism. The first option was to decompress messages i n the Consumer
module, which w o u l d have to receive a FetchResponse and then
iterate through the response's message set a n d replace its content
accordingly. The second option was to decompress messages immediately
w h e n they are deserialised, which w o u l d provide a very straightforward
way to work w i t h compressed messages, independent o n a
the Consumer implementation and it w o u l d also copy the behaviour
of the official Kafka client written i n Java. For these reasons, the latter
option was chosen.
The first idea was to p u t the decompressing mechanism directly
into the D e s e r i a l i z a b l e instance of Message i n the following manner:
instance D e s e r i a l i z a b l e Message where
d e s e r i a l i z e = do
— original deserialisation mechanism
message <- . . .
r e t u r n $ i f isCompressed message
then decompress message
e l s e message
where
decompress : : Message ->[Message]
decompres = . . .
Unfortunately, this code w o u l d not type check, due to the definition
of the D e s e r i a l i z a b l e type class, w h i c h requires the type of
d e s e r i a l i z e to be Get Message and not Get [Message]. Hence the decompression
code had to be placed somewhere higher i n the messagerelated
type hierarchy. W e can see that MessageSetMember, a type
directly w r a p p i n g a message, is neither a viable option as its type is
defined using a single Message. B y going higher i n the type hierarchy,
29
5. SUPPORTING C O M P R E S S I O N
we reach MessageSet, w h i c h is defined using [MessageSetMember],
which allows one MessageSetMember to be expanded into several message
set members and such members could be appended to the original
list without breaking type compatibility.
The implementation was then complicated by one minor obstacle.
W h e n a message containing compressed content is encountered and
its content decompressed, the result is a byte string w h i c h can not be
deserialised using any existent instance of the D e s e r i a l i z a b l e type
class. This is caused by the implementation of the aforementioned
type class - its function d e s e r i a l i z e does not take any arguments, but
rather reads input from an underlying State monad and i n this case,
the byte string is not a part of the state. Because of this, a short n e w
function h a d to be implemented to convert the decompressed byte
string into a value of MessageSet. The function reads its input and
repeatedly converts it into a value of MessageSetMember until there is
no input left or until the deserialisation fails for one of the message
set members.
A s a result of this change, reading messages using M i l e n a decompresses
previously gzipped messages automatically. It should
be mentioned, that this implementation supports only one level of
compression and if a hypothetical message A was compressed into
message B, and B was compressed into C , Milena w o u l d not be able to
reconstruct the original message A directly, but it w o u l d return only
message B.
Compressing Messages
The basic requirement for compression is the ability to compress any
number of messages, which requires the library to be able to compress
a value of MessageSet.
There were two obvious choices for placing the compressing mechanism.
It could have found its place i n the Producer module, making
compression dependent o n the Producer implementation, or it could
have found its place i n the serialisation mechanism. The second approach
w o u l d be independent on the producer implementation and
w o u l d be coherent w i t h the implementation of decompression. For
these reasons, the compressing mechanism was p r o g r a m m e d into
Milena as a part of the serialisation process.
30
5. SUPPORTING C O M P R E S S I O N
Since a message set is the compression unit i n Apache Kafka,
placing the compressing algorithm into the S e r i a l i z a b l e instance
of MessageSet emerged naturally, requiring a message set to signal
whether it should be compressed or not. The initial approach was
to add a new data constructor - CompressedMessageSet - bearing i n formation
about its compression codec. However, after a d d i n g this
new data constructor, I realised that the MessageSet type was an i n ternal
type of Milena and it should be possible to tamper with. Therefore
the compression information was simply added into the original
MessageSet data constructor, resulting i n the following declaration:
data MessageSet = MessageSet
_codec : : CompressionCodec,
_messageSetMembers : : [MessageSetMember]
} d e r i v i n g (Show, Eq)
To complete the implementation, a value of type MessageSet needed
to be compressed according to the indicated compression codec during
serialisation. The compression mechanism was placed into the
S e r i a l i z a b l e instance of type MessageSet and it consists of the following
steps:
1. It serialises the set's messages into a valid message set.
2. It compresses the byte string representing this message set.
3. It wraps the compressed byte string i n a Message, setting its
A t t r i b u t e s to reflect the used compression codec.
4. It creates a new message set containing the newly created mes-
sage.
5. It serialises the new set regularly.
The last step was to a d d a higher-level A P I , w h i c h w o u l d allow the
Producer module sending compressed messages. That was achieved
by a d d i n g a function produceCompressedMessages following the example
of the original produceMessages function:
31
5. SUPPORTING C O M P R E S S I O N
produceCompressedMessages : : Kafka m =>
CompressionCodec
-> [TopicAndMessage]
-> m [ProduceResponse]
produceCompressedMessages = . . .
A s a result of these changes, Milena is able to produce also compressed
messages.
Tests
The compression features were first tested manually b y setting u p a
Kafka broker, to which an official console producer and consumer were
connected. Code written using Milena was able to send compressed
messages i n such manner, that the console consumer properly showed
the original content of the sent messages, and the code was also able
to properly read the content of messages compressed by the console
producer.
O n top of the manual tests, two automatic integration tests were
added to the test suite. They test that messages compressed by Milena
are accepted by a Kafka broker, and that they can be later fetched and
contain their original content.
Possible Improvements
Although the introduced changes added support of Gzip compression,
further improvements could be done. For example, the implementation
does not allow configuring compression attributes - compression
level, buffer size etc. - even though the ability to set these could i m prove
the overall performance. Another useful feature w o u l d be automatic
batching of messages. A Producer could be programmed not to
send every message immediately, but buffer it and then compress and
send messages i n batches. The official Java client for Apache K a f k a
has similar feature [2].
A s for the code itself, there is a weak spot i n having compression
codec constants located i n two places, rather than just one and merging
them w o u l d definitely improve the quality of the code.
32
6 Supporting Multiple Versions of Requests
The K a f k a Protocol acknowledges the need to change over time and
enables a backwards compatible evolution of its A P I [3]. The versioning
mechanism is straightforward - every request/response pair exists
in a certain version and every request contains information about its
version and thus also information about the expected response [3]. For
instance, w h e n K a f k a receives a Fetch Request of version 0, it w i l l
always respond with a Fetch Response of version 0. W h e n it receives
a request of version 1, it w i l l always respond w i t h a response of version
1 etc. To introduce a new version of any request/response pair thus
means a d d i n g a new version of the pair without having to remove
the support of any existing versions. This way, Kafka can change only
some parts of its protocol and does not have to break compatibility
for its clients.
U p g r a d i n g K a f k a and taking advantage of its new features can
bring performance improvements, w h i c h creates a strong incentive
for M i l e n a to support communication over the newest version of the
protocol. In general, there were two options to approach this. Either
M i l e n a could drop support of the older versions, or it could extend
its code to support multiple versions. D r o p p i n g support of the older
versions w o u l d be inherently simple, as it w o u l d mean only replacing
the types of request/response pairs with their newer versions specified
in Kafka Protocol Guide and A Guide To The Kafka Protocol, but it would
also mean d r o p p i n g compatibility w i t h any code using Milena, as
the types w o u l d change. The second approach could be done i n a
completely backwards compatible fashion, but for the price of being
complex, as M i l e n a d i d not support any request versioning.
The initial choice was to go w i t h the first option, but after the
question of backwards compatibility arose, the choice was reconsidered
and alternatives allowing backwards compatibility were explored.
Based on a suggestion of Ixperta, the capabilities of phantom types and
Haskell language extensions GADTs and TypeFamilies were studied.
Using these, it w o u l d be possible to develop a backwards compatible
structure of requests and responses, w h i c h w o u l d allow communication
w i t h Kafka i n different versions of the protocol.
33
6. SUPPORTING M U L T I P L E V E R S I O N S O F REQUESTS
However, the code p u r s u i n g this option was rather complicated
and presented some naming challenges, especially i n Consumer and
Producer modules a n d thus it was decided that more elegant code
was preferred, even if it w o u l d slightly break Milena's backwards
compatibility A s a result of this decision, some intermediary layers
were removed, some functions were replaced by type class functions
and the TypeFamilies extension d i d not have to be used at all. This
chapter first introduces some general concepts/language extensions
related to the final solution of supporting multiple versions of requests
and then the solution itself.
6.1 Phantom Types
"Classical phantom types are datatypes in which type constraints are expressed
using type variables that do not appear in the datatype cases themselves.
" - Cheney; Hinze [14, p. 1]
We can see an example of a Haskell phantom type i n the following
piece of code:
newtype Request r e q resp = Request r e q
Here, Request is a phantom type, because the resp parameter does
not appear i n the definition of the Request data constructor. W h e n
used i n type definitions, a phantom type may be constricted (but does
not have to be):
foo = Request r e q resp -> S t r i n g
foo = . . .
bar : : Request r e q S t r i n g -> S t r i n g
bar = . . .
baz : : D e s e r i a l i z a b l e resp => Request r e q resp -> S t r i n g
baz = . . .
W h e n a phantom type is constricted, it allows the type system to
discover invalid uses, w h i c h then fail to type check. In the following
example, passing water to a c i d w o u l d fail because of incompatible
types:
34
6. SUPPORTING M U L T I P L E V E R S I O N S O F REQUESTS
water : : Request r e q B o o l
water = . . .
a c i d : : Request r e q S t r i n g -> S t r i n g
a c i d = . . .
It c o u l d be said that a value of a phantom type has some context,
according to w h i c h the value can be treated. For example, a request
may specify the format of the expected response.
6.2 GADTs
G A D T s is one of the standard Haskell language extensions and they
have to be explicitly enabled, for example by using a standard LANGUAGE
pragma1
:
{-# LANGUAGE GADTs #-}
The name is an abbreviation for Generalised Algebraic Datatypes a n d
the extension expands the options a n d syntax of defining data constructors
[7].
W h e n enabled, G A D T s allow data constructors to specialise their
type by using G A D T s syntax for defining a type [7]:
data Request r e q where
RequestVO : : RequestVO -> Request RequestVO
RequestVl : : RequestVl -> Request RequestVl
This declaration introduces a type Request r e q w i t h two data constructors.
Data constructor RequestVO accepts one parameter of type
RequestVO and returns a value whose type is specialised to Request
RequestVO rather than just Request req. Similarly, data constructor
RequestVl accepts one parameter of type RequestVl a n d returns a
value of a specialised type Request RequestVl rather than just Request
req.
Data constructors defined using G A D T s can be pattern-matched
normally2
:
1. GADTs had been already used i n Milena before the implementation of this
thesis, and thus did not have to be added.
2. Moreover, pattern matching on GADTs data constructors makes type constraint
context available to the right hand side of the match [7].
35
6. SUPPORTING M U L T I P L E V E R S I O N S OF REQUESTS
sendRequest : : Request r e q -> 10 ()
sendRequest (RequestVO r ) = . . .
sendRequest (RequestVl r ) = . . .
Both functionalities, the ability to specialise data constructor type and
pattern matching o n these constructors, are used w h e n defining a
generic request and working w i t h it, as described i n section 6.3.
6.3 Creating Generic Request Types
Generic Type Template
By using G A D T s and phantom types, a generic type for a request of
any version can be defined as:
newtype SomeRequestVO = . . .
newtype SomeResponseVO = . . .
newtype SomeRequestVl = . . .
newtype SomeResponseVl = . . .
data SomeRequest r e q resp where
SomeRequestVO : : SomeRequestVO
-> SomeRequest SomeRequestVO SomeResponseVO
SomeRequestVl : : SomeRequestVl
-> SomeRequest SomeRequestVl SomeResponseVl
SomeRequest is a phantom type, as the resp parameter is not a parameter
of any data constructor. G A D T s are then used to specialise this resp
parameter, based on the data constructor. A value of the SomeRequest
req resp type bears all the information needed for sending a request
to Apache Kafka and interpreting its answer. The request i n a specific
version is present i n the value, and the response version is present i n
the type of the value, w h i c h allows us to write:
send : : SomeRequest r e q resp -> resp
— The type system is able to infer that the type of resp
— should be SomeResponseVO and checks this constraint
send (SomeRequestVO r ) = . . .
— The type system is able to infer that the type of resp
— should be SomeResponseVl and checks this constraint
send (SomeRequestVl r ) = . . .
3 6
6. SUPPORTING M U L T I P L E V E R S I O N S OF REQUESTS
Renaming Existing Types
To a d d a generic request to every request/response pair existing i n
M i l e n a , the existing types and their data constructors h a d to be renamed
to keep naming consistent and avoid name collisions. Hence
a suffix V O was added to every request and response type and their
data constructors and code such as:
newtype FetchRequest = FetchReq ...
newtype FetchResponse = FetchResp ...
changed into:
newtype FetchRequestVO = FetchReqVO ...
newtype FetchResponseVO = FetchRespVO ...
Adding Generic Requests
After the types had been renamed, request types following the generic
template were added for every request/response pair existing i n
Milena.
Integrating Generic Requests into Milena
By itself, adding a request following the introduced template d i d not
a d d any functionality, it only added lines of code and further steps
had to be taken to integrate these generic requests.
The initial concept was to simply extend the existing code to work
w i t h generic requests and although the concept proved viable, allowing
finishing the implementation, another possible approach emerged
from it - the RequestMessage type could be replaced by as type class
generalising request types completely. If this type class specified functions
to retrieve a request's A P I key, its version and serialised representation,
introducing a new request to Milena w o u l d become only a
matter of creating an instance of this type class for the n e w request.
Whereas without such type class, adding a request w o u l d mean also
changing data types ReqResp and RequestMessage and also functions
a p i V e r s i o n , apiKey and doRequest, all defined i n the Protocol m o d ule.
For this reason, this approach was chosen, the type class was
named RequestMessage and its declaration along w i t h its usage can
be seen i n the following piece of code:
37
6. SUPPORTING M U L T I P L E V E R S I O N S O F REQUESTS
c l a s s RequestMessage r where
apiKeyValue : : r -> I n t l 6
apiVersionValue : : r -> I n t l 6
s e r i a l i z e R e q u e s t : : r -> Put
instance S e r i a l i z a b l e r e q => RequestMessage (
SomeRequest r e q resp
) where
apiKeyValue = 0
s e r i a l i z e R e q u e s t (SomeRequestVO r ) = s e r i a l i z e r
apiVersionValue SomeReqeustVOO = 0
Some functions of M i l e n a h a d to be changed accordingly, but the
modifications were trivial o n the whole, entailing mostly a change of
the type signatures of functions previously working w i t h the original
RequestMessage type.
6.4 Support of New of Protocol Constructs
A t this point, it was possible to implement some n e w versions of
requests a n d responses, but not all of them, because M i l e n a d i d not
support all the primitive types f r o m the K a f k a Protocol. Namely, it
could not w o r k w i t h values of BOOLEAN, nor NULLABLE_STRING a n d
neither it was able to properly (de)serialise sextuples.
Implementation of B O O L E A N Values
The biggest issue w i t h supporting BOOLEAN values was the lack of
documentation. A s of spring 2017, its representation i n the K a f k a
Protocol was not described i n neither Kafka Protocol Guide [3], nor A
Guide To The Kafka Protocol [4]. Albeit the intuition was that TRUE would
serialise into one byte containing 1, and FALSE into one byte containing
0, it had to be verified i n the source code of Apache Kafka [15]. Then,
the implementation of S e r i a l i z a b l e a n d D e s e r i a l i z a b l e instances
was trivial, as it only meant translating Boolean values into ones or
zeroes and vice versa.
38
6. SUPPORTING M U L T I P L E V E R S I O N S OF REQUESTS
Implementation of N U L L AB LE_STRING
A s suggested i n section 2.3, NULLABLE_STRING differs from STRING
minimally and therefore it was natural to design this data type to
mirror the Kaf k a S t r i n g data type. The Maybe m o n a d was then used
to wrap the string value to differentiate between a string of arbitrary
length and a NULL value, resulting i n the following data type:
newtype K a f k a N u l l a b l e S t r i n g = KNString {
_KNString : : Maybe B y t e S t r i n g
} d e r i v i n g . . .
Because the byte string value is wrapped i n Maybe type, the instances
of S e r i a l i z a b l e and D e s e r i a l i z a b l e could not be derived and had
to be implemented manually.
To completely follow the example set by Kaf k a S t r i n g , the new data
type needed an instance of the I s S t r i n g type class from the OverloadedStrings
extension and, because of the Maybe type, the instance could
not be derived automatically. M a n u a l implementation presented a
slight design issue - should an empty string be converted into Nothing,
or into Just "" ? In the end, the latter was chosen, resulting i n the
following I s S t r i n g instance:
instance I s S t r i n g K a f k a N u l l a b l e S t r i n g where
fromString x = KNString (Just (fromString x))
Choosing to convert an empty string into Just "" still allows passing
Nothing by using the data constructor - KNString Nothing - and does
not make any assumptions about an empty value meaning NULL. It
should be acknowledged that this might be only a matter of personal
taste and similar reasoning could be applied to the other option too.
Supporting Serialisation & Deserialisation of Sextuples
A d d i n g support for sextuple (de)serialisation meant adding instances
of the S e r i a l i z a b l e and D e s e r i a l i z a b l e type classes for a sextuple.
The implementation of the instances was mostly copied f r o m the i n stances
for a quintuple with one complication - the D e s e r i a l i z a b l e instances
for n-tuples use the l i f t M n functions from the C o n t r o l . Monad
module and the module does not export a l i f tM6 function, w h i c h
39
6. SUPPORTING M U L T I P L E V E R S I O N S O F REQUESTS
then had to be implemented manually. The implementation follows
the example set by l i f tM5 function as defined i n [16].
6.5 New Protocol Versions
After the preparatory steps described i n the previous parts of this
chapter, adding support for n e w versions of requests was trivial o n
the whole a n d mostly meant re-writing K a f k a Protocol definitions
into Haskell types. W i t h one exception being the Message structure,
whose support was more complicated and w h i c h is described i n the
following subsection.
Message VO, Message V I
W h e n version 2 of Fetch Request was added a n d integrated into
M i l e n a , the tests started failing for n o apparent reason. It turned
out that the protocol description i n Kafka Protocol Guide [3] was not
complete, and w h e n Fetch Request was upgraded to version 2, the
Message structure was also changed a n d Fetch Request of version 2
may return both versions of Message - version 0 and version 1 [4].
The different versions of messages are distinguished by the value of
their MagicByte field - value 0 suggests version 0, and value 1 suggests
version 1 [4]. Version 1 adds a field Timestamp, which should carry the
time of creation3
of the message i n milliseconds since the beginning
of the U n i x epoch.
One possible adaptation was obvious, a new data constructor could
be added for the Message type to represent version 1. This was i m plemented
a n d proved to be a viable solution. A s a part of this step,
the original data constructor Message was renamed to MessageVO, to
suggest versioning, resulting i n the following type declaration:
3. The meaning of the time value can be changed by manipulating the Attributes
field, but as of version 0.10.0 Apache Kafka recommends not using this feature and
hence this option is not discussed here in broader detail [4].
40
6. SUPPORTING M U L T I P L E V E R S I O N S OF REQUESTS
data Message = MessageVO { _messageFieldsVO : :
(Crc, MagicByte, A t t r i b u t e s , Key, Value)
}
I MessageVl { _messageFieldsVl : :
(Crc, MagicByte, A t t r i b u t e s , Time, Key, Value)
}
To be able to read messages i n the new format, the D e s e r i a l i z a b l e instance
of Message had to change. Based on the value i n the MagicByte
field it n o w returns data w r a p p e d i n proper data constructor and
similarly, the S e r i a l i z a b l e instance serialises either a quintuple w i t h
MagicByte equal to 0, or a sextuple w i t h MagicByte equal to 1.
The trickier part was i n passing current time to the MessageVl
data constructor. A l t h o u g h there is the standard timeA
package prov
i d i n g functions to access current time, they all w o r k w i t h i n the IO
monad, whereas the whole process of creating a value of Message
was pure. The first idea to approach this was to change the process to
automatically retrieve the current time while becoming impure. Albeit
an operational option, it was replaced after a consultation w i t h Peter
Trsko. The functions creating messages were changed to take current
time as their first parameter and a function currentTime was added
into the Producer module. This way, a Milena's user can decide h o w
to retrieve the current time, but also has the option to simply use the
provided function.
Finalising the Implementation
Once both versions of Message were supported, there was no other
obstacles and new request versions could be fully implemented. A s
a result of these changes M i l e n a works w i t h K a f k a Protocol supported
by Apache K a f k a 0.10.1. Furthermore, implementing a n e w
request/response pair i n Milena is n o w easier, as the generalisation
over requests was separated into a type class.
4. http://hackage.haskell.org/package/time
41
6. SUPPORTING M U L T I P L E V E R S I O N S OF REQUESTS
6.6 Tests
During the development, the functionality of added request versions
was tested by the existing test suite. However, these tests d i d not cover
sending multiple versions of a single request and therefore had to be
modified accordingly. A s the test cases were being added, code in the
test suite became rather bloated and was split into several modules.
Firstly, Kaf kaTest module was refactored out of the test suite. This
module contains functions shared by multiple tests and, most importantly,
a function which creates the underlying Kafka monad and a connection
to a Kafka broker. Secondly, the tests for message production
and consumption were separated into custom modules T e s t s . Consume
and Tests.Produce.
42
7 Supporting Groups
The concept of consumer groups makes parallelising of message processing
easier, providing the incentive for implementing this feature.
The Kafka Protocol defines several request/response pairs w h i c h
give a consumer full control over its group membership - a consumer
can join and leave any group at any time by issuing a proper request [3].
However, preparing a proper request can be complicated - as the A P I
defined by Kafka Protocol Guide [3] is very generic1
, some functionality
is left to be implemented by the client [4]. This thesis adds only
support of the generic form of group membership management as i m plementing
the consumer group functionality compatibly w i t h other
client libraries w o u l d require setting up a whole system of clients and
extensive testing of compatibility, w h i c h lies beyond the bounds of
this thesis.
7.1 Group Membership Management Process
The information about the Apache Kafka group membership management
process i n this section is based on Documentation of Apache Kafka
Group Membership Management [17].
The process of group membership management has several phases.
In the first phase consumers have to send a JoinGroup Request, which
determines the active members of a group. One of the active group
members is then randomly chosen to become the leader of the group,
w h i c h is indicated to the group member i n the JoinGroup Response.
Once the joining phase is completed, every group member has to
send a SyncGroup Request, i n whose response a consumer receives
its group-related state. The group state is determined by the group
leader, which sends it as the contents of its SyncGroup Request. After
finishing this step, an active group is established.
The members of an active group should periodically indicate their
activity be sending a Heartbeat Request. This enables further m a n agement
of the group - heartbeat responses are used to indicate that,
for instance, a new member has joined or that a member has left a
1. The aim is to support other use cases, not just consumer groups [4].
43
7. SUPPORTING G R O U P S
group. If a heartbeat request indicates a change i n the group state,
the members should react adequately as otherwise Kafka might stop
sending them data. If a group member does not send a heartbeat request
for too long, Apache Kafka w i l l consider it inactive and w i l l stop
communicating w i t h the consumer2
.
W h e n a consumer should leave a group, it can either leave its
session to expire or send a LeaveGroup Request. The latter is preferred,
as then Kafka immediately knows that a consumer was removed and
can adapt to the new situation faster [4].
7.2 Implementation
The implementation itself was straightforward, it was needed to add
request/response types for individual operations - joining and leaving
a group, syncing it and sending a heartbeat. Therefore request and
response types were added into the Protocol module and Consumer
module was updated to contain functions w r a p p i n g the individual
operations. Arguably, the added A P I could be further improved to
provide higher level access to the group management process, but to
do that properly, some feedback from using the implemented form
w o u l d be needed and I lack such experience.
7.3 Tests
The implementation of tests for group membership management was
complicated by the dependencies between the individual phases. First
a consumer must join a group, after which the group has to be synchronised
and then the consumer might send a heartbeat and/or leave the
group. If the operations happened i n different order, Apache Kafka
w o u l d always return an error.
There are several options to approach dependent tests, every option
having its benefits and drawbacks. One solution is to write only one
test case, which tests one phase after each other together in one, usually
long, piece of code. Second solution is to separate tested parts into
several functions and write several test cases, one for each tested phase,
2. The amount of time, after which a member is considered inactive, can be config-
ured.
44
7. SUPPORTING G R O U P S
w h i c h then call the separated functions as needed. Another possible
approach is to make the test cases share state and depend on each
other. After some consideration, the first and second approaches were
rejected - even though they are cleaner than the last option, their code
seemed complicated.
The third approach presented an issue i n sharing state between
the test cases. A t first, the StateT m o n a d [18] was explored, to serve
as c o m m o n state for i n d i v i d u a l test cases, but it was not possible to
combine it w i t h the used testing framework. Later I was advised by
Ixperta to use module Data. IORef which provides functionality similar
to overwriting a variable [19]. U s i n g this module, it was possible
to write tests i n the following manner:
t e s t s = do
sharedData <- runIO $ newIORef (...)
i t " j o i n s a group" $ do
writelORef sharedData (...)
i t "synces the group" $ do
(...) <- readlORef sharedData
A l t h o u g h the tested functions are a part of the Consumer module,
the tests were placed into a separate module Tests . Group, to avoid
overcrowding the Consumer module tests.
45
8 Conclusion
The thesis managed to accomplish all of its m a i n goals. The concept
of lenses was described and the three key functionalities - support
of compression, consumer groups and K a f k a Protocol supported by
Apache K a f k a 0.10.1 - were successfully developed along w i t h their
tests. The code itself was reviewed by one of Ixperta's programmers
and received positive feedback.
Although the thesis successfully developed all the necessary functionality,
it possibly brought some weak spots. Namely, i n the compression
related code, constants are defined i n more than one place, which
is not a good practise and should be removed. In other instances, I lack
the knowledge of Haskell/functional paradigm best practises needed
to objectively assess the quality of the code.
The thesis also resulted i n several pull requests into the main repository
of Milena. Two of the p u l l requests h a d been already merged
before the thesis was finished, but others were opened only a few
days before the thesis was completed and their final status is thus
u n k n o w n . The process of accepting these p u l l requests might entail
further changes w h i c h could not be described i n the thesis itself.
In the future, this thesis could be expanded i n several ways. The
added compression functionality could be improved to support more
compression codecs and automatic batching of messages. The consumer
group management A P I could be expanded to support more
than just the generic form and the support of multiple versions of requests
could be leveraged to extend Milena's functionality i n general.
47
A Source Code
The data archive of this thesis contains a git repository of the m o d ified
version of M i l e n a 1
. The functionality is located i n three separate
branches - compression, versioned-requests and group-api. The
project uses Stack2
as the build tool. To run the tests, an Apache Kafka
broker must be accessible at localhost, o n port 9092.
1. The repository is also accessible from https://github.com/pavelkucera/
milena-1
2. https://docs.haskellstack.org/en/stable/README/
49
Bibliography
1. List of Projects Using Apache Kafka [online] [visited o n 2017-04-15].
Available from: h t t p s : //kafka.apache.org/powered-by.
2. Official Apache Kafka Documentation [online]. Apache Software Foundation
[visited o n 2017-04-15]. Available from: h t t p s : / / k a f k a .
apache.org/documentation.
3. Kafka Protocol Guide [online]. Apache Software Foundation [visited
on 2017-04-11]. Available from: h t t p s : / / k a f k a . apache . o r g /
p r o t o c o l .
4. A Guide To The Kafka Protocol [online]. Apache Software Foundation
[visited on 2017-04-11]. Available from: https : / / c w i k i . apache .
org / confluence / d i s p l a y / KAFKA / A + Guide + To + The + Kafka +
P r o t o c o l .
5. Documentation of the Data.Serialize.Get module [online] [visited on
2017-04-15]. Available from: h t t p : / / hackage . h a s k e l l . org /
p a c k a g e / c e r e a l - 0 . 5 . 4 . O / d o c s / D a t a - S e r i a l i z e - G e t . h t m l .
6. Documentation of the Data.Serialize.Put module [online] [visited on
2017-04-15]. Available from: h t t p : / / hackage . h a s k e l l . org /
p a c k a g e / c e r e a l - 0 . 5 . 4 . O / d o c s / D a t a - S e r i a l i z e - P u t . h t m l .
7. Documentation of GHC language extensions [online] [visited on
2017-04-20]. Available from: h t t p s : / /downloads . h a s k e l l . o r g /
~ g h c / l a t e s t / d o c s / h t m l / u s e r s _ g u i d e / g l a s g o w _ e x t s . h t m l .
8. S T E C K E R M E I E R , Albert. Lenses i n Functional Programming [online].
2015 [visited on 2017-04-25]. Available from: h t t p s : //www21. i n .
turn.de/teaching/fp/SS15/papers/17.pdf.
9. F I S C H E R , Sebastian; H U , Zhenjiang; P A C H E C O , H u g o . A clear picture
of lens laws —Functional Pearl—. In: Proceedings of the 12th
Conference on Mathematics of Program Construction (MFC 2015). 2015.
Available also from: h t t p : / / s e b f i s c h . g i t h u b . i o / r e s e a r c h / p u b /
Fischer+MPC15.pdf.
10. Overview of lens package functionality [online] [visited o n 2017-04-25].
Available from: h t t p s : / / github . com / ekmett / lens / w i k i /
Overview.
51
B I B L I O G R A P H Y
11. Lens package examples [online] [visited o n 2017-04-25]. Available from:
h t t p s : / / g i t h u b . c o m / e k m e t t / l e n s / w i k i / E x a m p l e s .
12. Separate part of the Apache Kafka documentation, related to compression [online].
Apache Software Foundation [visited on 2017-04-13]. Available
from: https : / / c w i k i . a p a c h e . o r g / c o n f l u e n c e / d i s p l a y / K A F K A /
Compression.
13. LIS Haskell 7.19 (ghc-8.0.1) package list [online]. F P Complete [visited
on 2017-04-13]. Available from: https : //www. stackage . o r g / l t s -
7.19.
14. C H E N E Y , James; H I N Z E , Ralf. First-class phantom types. 2003. Available
also from: h t t p : / / h d l . handle . net /1813/5614. Technical
report. Cornell University.
15. Commit adding boolean type to the Kafka protocol [online]. Apache
Software Foundation [visited o n 2017-04-15]. Available
from: https : / / github . com / apache / kafka / commit /
33d745e2dcfa7a9cac90af5594903330ad774cd2.
16. Source code of Haskell UftM functions [online] [visited o n 2017-04-15].
Available from: h t t p : / / h a c k a g e . h a s k e l l . o r g / p a c k a g e / b a s e -
4 . 9 . 1 . 0 / d o c s / s r c / G H C . B a s e . h t m l # l i f t M .
17. Documentation of Apache Kafka Group Membership Management [online]
[visited o n 2017-04-15]. Available from: https : / / c w i k i .
apache . org / confluence / d i s p l a y / KAFKA / Kafka + C l i e n t -
side+Assignment+Proposal.
18. Control.Monad.Trans.State.Strict Module Documentation [online] [visited
o n 2017-04-15]. Available from: https : / /hackage . h a s k e l l .
o r g / p a c k a g e / t r a n s f o r m e r s - 0 . 5 . 4 . 0 / d o c s / C o n t r o l - M o n a d T
r a n s - S t a t e - S t r i c t . h t m l .
19. Documentation of Data.IORef module [online] [visited o n 2017-04-20].
Available from: https : / /hackage . h a s k e l l . o r g / p a c k a g e / b a s e -
4 . 9 . 1 . 0 / d o c s / D a t a - I 0 R e f . h t m l .
52