Update readme v1.0.0 mk

[klippx]

Comments for @tulios

• How does one implement a custom decoder for consuming?
• Elaborate on the statement "It will also provide more utility functions to give your code more flexibility" for eachBatch
• In resolveOffset I understand that the default behaviour is that processed messages within a batch will be commited in case of errors, while the rest will remain unprocessed. But it is not clear to me why I would like to change this. What is the use case? It is only adding confusion for me.
• Seek: you don't need to await consumer#run - can you elaborate why this is the case?
• Pause & Resume: I removed mentions of "unsupported by the library"
• Pause & Resume: The example is very weird. Maybe its fine, but its not the way ppl use topics. Maybe use 1 topic and pause because of a dependency being down which seems more like a useful scenario?

[Nevon]

This isn't really true though. eachMessage is essentially a default implementation of eachBatch. We still consume a batch at a time from the brokers, and the offset is only resolved, not committed, on each message.

Situations where you might want to use eachBatch is where you need to do something with all messages at once. For example, maybe you want to send some data from each message to a remote batch API, so that for N messages you call the remote API once.

[Nevon]

The example is very weird. Maybe its fine, but its not the way ppl use topics. Maybe use 1 topic and pause because of a dependency being down which seems more like a useful scenario?

We actually do something similar, where we expose an HTTP endpoint to pause/resume. When I wrote the example, I just didn't want to make assumptions about how the pause/resume would be invoked, so invoking it via a Kafka message seemed appropriate. 😅

If you have an idea for a clearer example, I'm all for it. I just can't think of one that wouldn't involve more non-KafkaJS related cruft. For example, if we take the example of your dependency responding with 429:

await consumer.connect()

await consumer.subscribe({ topic: 'jobs' })

await consumer.run({ eachMessage: async ({ topic, message }) => {
  try {
    await sendToDependency(message)
  } catch (e) {
    if (e instanceof TooManyRequestsError) {
      consumer.pause([{ topic }])
      setTimeout(() => consumer.resume([{ topic }]), e.retryAfter * 1000)
    }

    throw e
  }
}})

[tulios]

I like the detailed explanation but can we move it to the bottom? I think the beginning of the readme should provide a quick setup and the most common use case, detailed information such as the custom logger or how the retry mechanism works could live further down. WDYT?

[klippx]

definately, good point, it clutters up too much

[tulios]

❤️

[tulios]

@klippx

How does one implement a custom decoder for consuming?

The codec works for both consumers and producers; the consumer will detect that the message is compressed with some codec and it will use the appropriate decompress function do decompress. If the codec is implemented in KafkaJS, it'll just work.

Elaborate on the statement "It will also provide more utility functions to give your code more flexibility" for eachBatch

The eachBatch functions receive 3 helpers besides the batch: resolveOffset, heartbeat,
and isRunning.

In resolveOffset I understand that the default behavior is that processed messages within a batch will be committed in case of errors, while the rest will remain unprocessed. But it is not clear to me why I would like to change this. What is the use case? It is only adding confusion for me.

When we shut down the consumer, it will wait for the full batch to be processed before it exit, if you use isRunning to stop processing the messages the consumer will automatically commit the last offset, skipping the messages you didn't consume. If you make sure you call resolveOffset and you disable the auto-resolve, you can quickly shut down the consumer without losing/skipping any messages.

Seek: you don't need to await consumer#run - can you elaborate why this is the case?

It's possible to optionally await consumer.run, this will make sure your promise resolves when the consumer successfully receives its first batch, this isn't necessary since this operation will happen multiple times throughout the app lifecycle but it gives an extra check. If you call seek after waiting for run, it means that the consumer received the first batch and then performed the seek operation. To make sure your consumers start from the seek definition you have to call run without await.

The reason for all of that is that we need to initialize the consumer before we invoke the seek operation and run is the best place to initialize the consumer.