Errata

In Chapter 4 --> Commits and Offsets --> Commit Current Offset , 2nd paragraph.

"By setting auto.commit.offset=false, offsets will only be committed when the application explicitly chooses to do so."

"auto.commit.offset" should be "enable.auto.commit" ?

Note from the Author or Editor:
(This is page 77 of the PDF).

The comment is correct: "auto.commit.offset" should be "enable.auto.commit"

"...handling X messages per second and a data rate of X megabits per second..." looks like placeholder values. Other errata mentions this also appearing in "section 710" of the kindle edition but I'm confirming here that it's in the pdf on page 30.

Note from the Author or Editor:
Thanks for the note. This has been resolved for the next update

"Otherwise, ephemeral storage (such as the AWS Elastic Block Store) might be sufficient." AWS EBS is not ephemeral. Perhaps rewrite as "Otherwise, ephemeral storage or AWS EBS could be used depending on the level of replication available and resilience required."

Note from the Author or Editor:
The word that should have been used here is "remote". This will be resolved in the next update

A variable "kafkaProps" is defined as a local variable with modifier "private". An access modifier is not allowed here. It should be removed.

A variable "producer" is initialized, but it is not definied. At P.57, "producer" is initialized with variable declaration. It can replace with:
KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProps);

Note from the Author or Editor:
1. That depends on context, but I agree it can be confusing in a snippet.

We need to replace:
"private Properties kafkaProps = new Properties();"
with:
"Properties kafkaProps = new Properties();"
on page 44.

2. I don't see a problem in page 57 example.
We create the producer with "Producer<String, Customer> producer = new KafkaProducer<String,
Customer>(props);" and the "props" instance is created in the first line of the example.

"... the largest message you can send is 1MB or the producer can batch 1,000 messaged of size 1 K each into one request."

1. it should be 1024 instead of 1000
2. it should be "1KB each" (missing "K" and no space -- compare 1MB")

The statement

ProducerRecord<String, Customer> record = new ProducerRecord<>(topic, customer.getId(), customer);

uses "customer.getId()" as the key where "id" is defined to be an integer in the schema.
This in contrast to the type definition "ProducerRecord<String, Customer>" that requires a String key.

Using customer.getName() would align the code with the schema.

Note from the Author or Editor:
Correct. customer.getId() in the example should be replaced with customer.getName()

String email = "example " + nCustomers + "@example.com"

Missing final semicolon:
String email = "example " + nCustomers + "@example.com";

customer.put("id", nCustomer);

Should be
customer.put("id", nCustomers);
since that is how the variable is named in the rest of the code sample.

The first ProducerRecord<Integer, String> should be ProducerRecord<String, String> to match the arguments.

The second ProducerRecord<Integer, String> should be ProducerRecord<String> to match the arguments.

Note from the Author or Editor:
It will be ProducerRecord<String, String> in both examples, both type arguments are required even when the key is null.

"...resulting in one partition being about twice as large as the rest. " Hard to see how twice as large is derived, Perhaps just say "much larger".

Note from the Author or Editor:
Good point.

Lets change:
"...resulting in one partition being about twice as large as the rest. "
to
"...resulting in one partition being much larger than the rest."

The return value for the "Banana" customer should be "numPartitions - 1" and not "numPartitions" because partitions are number starting at zero.

Note from the Author or Editor:
Good catch!

In the code example, where we say:
return numPartitions;

It should be:
return numPartitions - 1;

Figure 4-2: There is one consumer group with two consumers; not two consumer groups.

Figure 4-3: There is one consumer group with four consumers; not four consumer groups.

Figure 4-4: There are not more consumer groups than partitions; there are more consumers than partitions. Also, I don't think this means missed messages just idle consumers.

Figure 4-5: The reference to missing messages is confusing. Both consumer groups will get all messages.

Note from the Author or Editor:
The first 3 figure captions have been corrected to properly refer to a number of consumers in a single group. The last two figure captions have been corrected to not refer to missing messages.

The argument to subscribe is not a String but a Pattern. Thus

consumer.subscribe("test.*");

is wrong and should be

consumer.subscribe(Pattern.compile("test.*"));

Note from the Author or Editor:
Correct.

At the very top of the page:
consumer.subscribe("test.*");
should be replaced with:
consumer.subscribe(Pattern.compile("test.*"));

`custCountryMap` counts number of records with the record value as a key. Is `custCountryMap` a `Map<String, Integer>`?

The condition of the if-block in the code:

if (custCountryMap.countainsValue(record.value())) {
updatedCount = custCountryMap.get(record.value()) + 1;
}

Should it be

if (custCountryMap.containsKey(record.value())) {
updatedCount = custCountryMap.get(record.value()) + 1;
}

?

Additionally, that can be simplified with `getOrDefault()` method:

int updatedCount = custCountryMap.getOrDefault(record.value(), 1);

Note from the Author or Editor:
Good catch. This is correct.

Lets replace:
if (custCountryMap.countainsValue(record.value())) {

With
if (custCountryMap.countainsKey(record.value())) {

...(determined by max.message.size...

should be max.message.bytes

Note from the Author or Editor:
`max.message.size` should be `message.max.bytes` (since I refer to the broker config here).

max.message.bytes is the topic level max, so the largest message should come from message.max.bytes if you are referring to broker.

However it might be worth mentioning both topic and broker configs here?

Note from the Author or Editor:
Lets replace `max.message.size` with `message.max.bytes`.
This doesn't look like the right place to get into topic configuration.

The default for session.timeout.ms is 10s, not 3 - 3 is for heartbeat.interval.ms

Note from the Author or Editor:
Good catch. Should be:
"The amount of time a consumer can be out of contact with the brokers while still considered alive defaults to 10 seconds."

public void onComplete(Map<TopicPartition,
OffsetAndMetadata> offsets, Exception exception) {
if (e != null)
log.error("Commit failed for offsets {}", offsets, e);
}

Exception is called exception in the method signature but e in method body.

Note from the Author or Editor:
This is indeed a mistake.

Should be:
"OffsetAndMetadata> offsets, Exception e) {"

Quote:

"has offset 5000, you can call commitSync() to commit offset 5000"

If offset 5000 was consumed, one should commit offset 5001.

Note from the Author or Editor:
Correct.

Lets change:
"If you are in the middle of processing a batch of records, and the last message you got from partition 3 in topic “customers” has offset 5000, you can call commitSync() to commit offset 5000 for partition 3 in topic “customers.”"
To:
"If you are in the middle of processing a batch of records, and the last message you got from partition 3 in topic “customers” has offset 5000, you can call commitSync() to commit offset 5001 for partition 3 in topic “customers.”"

seekToBeginning(TopicPartition tp) and seekToEnd(TopicPartition tp)

Should this be
seekToBeginning(Collection<TopicPartition> tp) and seekToEnd(Collection<TopicPartition> tp)?

Note from the Author or Editor:
Correct.

currentOffsets.put(new TopicPartition(record.topic(), record.partition()), record.offset());

Should it be

currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
new OffsetAndMetadata(record.offset()+1) ;

?

>> Now the only problem is if the RECORD is stored in a database and not in kafka...

should be changed to

>> Now the only problem is if the OFFSET is stored in a database and not in kafka...

The error is indicated in caps

Note from the Author or Editor:
This comment is correct. Instead of "Now the only problem is if the record is stored in a database and not in kafka" page 85 should say "Now the only problem is if the offset is stored in a database and not in kafka".

1) Should the check `data.length` check for 16? 8 bytes for `id` and 8 bytes for `nameSize` ?

2) Exception message does not make sense:

"Size of data received by IntegerDeserializer is shorter than expected."

There is no IntegerDeserializer, but a CustomDeserializer is implemented.

3) is the check for `null` and `data.length` required/intended? It's all wrapped with a try-catch-block anyway.

Note from the Author or Editor:
1. Good point.
"if (data.length < 8)"
should be
"if (data.length < 16)"

And "IntegerDeserializer" should be "deserializer".

2. The code example here could definitely be better. But this is outside the scope of errata and this is an example of what one shouldn't do.

consumer.subscribe("customerCountries")

Should it be

consumer.subscribe(Collections.singletonList("customerCountries"))
?

props.put("value.deserializer", "org.apache.kafka.common.serialization.CustomerDeserializer");

Package org.apache.kafka.common.serialization should not be used for custom code.
A simpler solution might be
props.put("value.deserializer", CustomerDeserializer.class.getName());

Note from the Author or Editor:
Good catch.

String nameSize = buffer.getInt();

Variable nameSize was already defined at the top of the method. Besides it should have type int (not String). The line should be replaced with
nameSize = buffer.getInt();

Note from the Author or Editor:
Corrected "String nameSize = buffer.getInt();" to "nameSize = buffer.getInt();"

name = new String(nameBytes, 'UTF-8');

Java strings are delimited by double quotes. It should be
name = new String(nameBytes, "UTF-8");

record.value().getId()

Should be

record.value().getID()

since this is how the method was defined for class Customer on page 89

Note from the Author or Editor:
The typo "record.value().getId()" has been fixed to "record.value().getID()"

The for loop straddling across page 90 and page 91 should be followed (on page 91) by a commit statement like

consumer.commitSync();

Note from the Author or Editor:
Correct.

At the very top of page 91 there are 4 lines that read:
record.value().getId() + " and
current customer name: " + record.value().getName());
}
}

It should read:
record.value().getId() + " and
current customer name: " + record.value().getName());
}
consumer.commitSync();
}

(Note that the new line should be indented to align with the curly bracket just above it).

The property keys "key.serializer" and "value.serializer" should be

"key.deserializer" and "value.deserializer"

Note from the Author or Editor:
Correct.
In this example
"key.serializer" should be "key.deserializer"
and
"value.serializer" should be "value.deserializer"

As others have pointed out the props keys for the deserializers are wrong (in the example, the props for the serializers are uses). Furthermore, to get it working, I had to specify "specific.avro.reader" to true otherwise I get a ClassCastException from GenericAvroRecord to my generated class. I believe that property should be always set to true when deserializing avro messages?

thank you for the fantastic work on this book. It's great!

Note from the Author or Editor:
I think this comment is for an older print of the book. Note that the errors pointed out here are for page 90 (not 91) in the PDF.

The deseralizers are correct in the recent version, but the reader is correct that a property is missing.

Lets add:
props.put("specific.avro.reader","true");
In the line after:
props.put("value.deserializer",
"io.confluent.kafka.serializers.KafkaAvroDeserializer");
At the code example on the end of page 90

> Now replica 3 is unavailable and replica 0 is back online. Replica 0 only has messages 0-100 but not 100-20

In this section, assume that there are three replicas of a partition. In this paragraph, about unclean leader election is discussed. So the unavailable replica must be replica 2 but not replica 3.

Note from the Author or Editor:
The description in the error report is correct. It should be "Now replica 2 is unavailable" instead of "replica 3".

"In conjunction with the min.insync.replica configuration..."

The configuration parameter should be min.insync.replicas. The final 's' is missing

Note from the Author or Editor:
Correct, this should be "min.insync.replicas"

The paragraph references configuration properties
key.converter.schema.enable
and
value.converter.schema.enable

Should it be

key.converter.schemas.enable
and
value.converter.schemas.enable

?

Note from the Author or Editor:
Yes. Should be:
key.converter.schemas.enable
and
value.converter.schemas.enable

The following code does not work :

echo '{"name":"load-kafka-config", "config":{"connector.class":"FileStream-
Source" ...

instead the connector.class should be FileStreamSource :

echo '{"name":"load-kafka-config", "config":{"connector.class":"FileStream
Source" ...

My 2 cents ...

Note from the Author or Editor:
Correct. This error repeats twice. In the code block starting with "echo" and the code block immediately below. I believe this is the result of breaking a single word "FileStreamSource" into two lines. Since the result ("FileStream-Source") indeed doesn't work, I suggest breaking the line after the preceding column (i.e. after "connector.class":) in those two code blocks.

In command

bin/kafka-console-consumer.sh --new --bootstrap-server=localhost:9092 --topic kafka-config-topic --from-beginning

option --new is invalid. It should probably be --new-consumer but since version 0.10.1 it is no longer needed.

Note from the Author or Editor:
Corrected to --new-consumer in both places in the chapter.

In
"choosing the lower of max.tasks configuration and the number of tables"

Should the configuration property be "tasks.max" instead?

Reported lag should be 5 instead of 4.

The committed offset is the next offset to be consumed. Thus, if offset 3 is committed, messages from offset 3 to 7 needs to be mirrored, and thus lag is 5.

Note from the Author or Editor:
Good catch!
We need to replace:
"In the diagram, the real lag is 2, but the
kafka-consumer-groups tool will report a lag of 4 because MirrorMaker
hasn’t committed offsets for more recent messages yet."
With:
"In the diagram, the real lag is 2, but the
kafka-consumer-groups tool will report a lag of 5 because MirrorMaker
hasn’t committed offsets for more recent messages yet."

There are two occurrences of "fetch.max.wait"
It should probably be
"fetch.max.wait.ms"

Entry:

- log-end-offset: I think that the log-end-offset is one larger. If a topic is empty, log-end-offset should be zero. If there is five produced messages (from offsets 0 to 4) log-end-offset should be 5, but not "the offset of the last message produced" that would be 4.

Question:

- owner: is this the same as `client.id`? If not, what does "provided by the group member" mean?

The paragraph mentions options --new-consumer and --broker-list.
The former is no longer required as of Kafka 0.10.1.
The latter is actually called --bootstrap-server and allows to specify the Kafka broker to connect to.

Note from the Author or Editor:
"broker-list" has been fixed to "bootstrap-server". The "new-consumer" flag is left, however, as it is required for previous versions.

In the paragraph the "kafka.coordinator.GroupMetadataManager$OffsetsMessageFormatter"
is quoted, but first:
the escape character '\' must be used like:

kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter


And then the class package is no longer valid (for new Kafka releases) is now:

kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter

In addition 'zookeeper' parameter is no longer valid, I propose to replace it with 'bootstrap-server' for example:

bin/kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server localhost:9092,localhost:9093,localhost:9094 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --consumer.config ./config/consumer.properties

Note from the Author or Editor:
The example uses single quotes, which does not require escaping the dollar sign. I have added a note that the class name is different in 0.11. The option for using --bootstrap-server vs. --zookeeper is covered elsewhere in the chapter. As of 0.11, --zookeeper is supported, and this book does not cover Kafka 1.0.

In

"It has enough messages to fill a batch based on the max.partition.bytes configuration"

should the configuration property be "batch.size" instead?

Note from the Author or Editor:
The configuration name has been updated to reflect the new producer configurations.

According to the text description and the materialized view on the right side of the figure, the final row of the stream of events on the left side of the figure should be labeled "Sale" and not "Return".

Note from the Author or Editor:
Right!

The in the last two rows in the left table, it looks like both "blue shoes, 300" and "green shoes, 299" are "return" events, where "blue shoes, 300" is a return but "green shoes, 299" is a sale.

Such applications need to maintain state within the application because each event can be handled independently.
->
Such applications do not need to maintain state within the application because each event can be handled independently.

Note from the Author or Editor:
The feedback is accurate - should be "do not need" instead of "need".

This is in the kindle edition of the book -- "location 710", not page 710.

It's in the section "Chapter 4: Installing Kafka > Hardware Selection > Memory", and it looks like a placeholder value that was just not filled in later.

"Even a broker that is handing X messages per second and a data rate of X megabits per second can run with a 5GB heap."

My assumption is that those X's were intended to be replaced with numeric values of some sort :)

Note from the Author or Editor:
Thanks for the note! I've added the values that should have been there in an upcoming revision (committed)