Help needed. I'm losing messages

[matreyes]

I have a pipeline which could be explained as:

<topic_raw>
|> brod_group_subscriber_v2 (message_set) callback
|> processing (it takes 5-10 seconds) for batch
|> :brod.produce_sync (for each message)
|> <topic_processed>

I have created another application that counts the number of messages on topic_raw and topic_processed by hour, and I'm loosing ~ 10% of the messages, without seeing any warning on logs (I would expect "at least once").

The subscriber configuration is:

  group_config = [
    offset_commit_policy: :commit_to_kafka_v2,
    offset_commit_interval_seconds: 5,
    session_timeout_seconds: 20,
    rejoin_delay_seconds: 3,
    reconnect_cool_down_seconds: 10
  ]

config = %{
  client: @source,
  group_id: Application.get_env(:parser, :pipeline)[:group_id],
  topics: [Application.get_env(:parser, :pipeline)[:input_topic]],
  cb_module: __MODULE__,
  message_type: :message_set,
  group_config: group_config,
  consumer_config: [
    begin_offset: :earliest,
    min_bytes: 100_000,
    max_bytes: 1_000_000,
    max_wait_time: 5_000
  ]
}

:brod.start_link_group_subscriber_v2(config)

And the callback is something like this:

def handle_message(messages, _state) do
    topic_parsed = Application.get_env(:parser, :pipeline)[:output_topic]

    with trxs <- parse_messages(messages),
         :ok <- produce(trxs, topic_parsed)
      {:ok, :ack, nil}
    else
      {:error, e} ->
        Logger.error("Error producing: #{inspect(e)}")
        raise "Failing to produce messages"
    end
end

def produce(messages, topic) do
    for {:ok, id, d} <- messages do
      case :brod.produce_sync(@sink, topic, :hash, d.key, d) do
        :ok ->
          :ok

        {:error, e} ->
          Logger.error("Error producing the message #{id} because #{e} ")
          raise "Error producing message"
      end
    end

    :ok
  end

I'm really clueless and urged. If you could help me, I would really appreciate It ❤️

[zmstone]

Hi

How are the messages counted ?
Are the topics compacted ?

[zmstone]

By as metadata, I mean for example upstream partition and offset as message headers for downstream

[matreyes]

I love the Idea of putting the original offset as metadata! I will do that. I really needed a smart advice!

[matreyes]

Well, they are not compacting, but I will however do what you recommended and I will check if offsets increase monotonically. If not, any other idea?. I suppose that no offset is commited without the :ack callback, I'm right?

[zmstone]

No. The consumer should not commit offsets without the :ack return.
Which Kafka version is it?
And producer config?
Possible reasons:
message parser function has dropped some messages
Processor app is not consuming one of the upstream partitions
Counter app is not consuming one of the downstream partitions

[matreyes]

We use Kafka 1.1
1.- Message parser function has dropped some messages: We have a "dead letter" and logs for that case. I will review It anyway.
2.- Specific partition not being consumed: actually when we do the counting, we use the ID/key and hour for grouping. So If it was a partition, I would expect the counter to be on 0 for a specific key. But It's always ~10% less.
So, I will approach it with the offset on metadata and re-checking the possibility that I'm functionally rejecting some messages (some bug). And I will not take more of you valuable time (I really appreciate your work for this community ❤️ )

Thank you very much!

[matreyes]

Hi @zmstone. After some time digging I've realized that our Kafka cluster (v1.1) is not registering the consumer group.
It's very very strange. In Staging and Dev I don't have any issue, so I can't reproduce, but in production if we list the consumer groups
(sh kafka-consumer-groups.sh --bootstrap-server localhost:9092 --all-topics --list), it doesn't appear; if I create one with kafkacat, it does!

As i'm using [begin_offset: :earliest] every time it starts from the beginning :(

Do you think it could involve kafka_protocol?

Pretty similar:
https://stackoverflow.com/questions/49319885/sometimes-a-new-consumer-group-does-not-work

Probably I will try creating the group with kafkacat and then just joining, similar to the stackoverflow solution

Best

[matreyes]

Actually I've changed it to earliest, Before it was on latest. And yes now I have days of duplicates :( .
I keep investigating. Best!

[matreyes]

Interesting @zmstone

We do long time processing (we use message_set, and we call APIs), so it could be causing the difference between dev and prod.

session.timeout.ms is in 10 seconds (default), so, if i'm unable to heartbeat before that, could I loose the session before the first commit? and in that case consumer group id is not created?
But far as I can see, heartbeat is managed by the consumer_group_coordinator, so it shouldn't be blocked between polls.

There is no special config for _consumer_offsets topic :(

I'm asking to the kafka team to increase the session.timeout.ms, do you think it could help ?

Thanks again!

[zmstone]

Please also check the default offset retention: committed offsets have retention, if no new commit happened to a partition for long enough time, the offset commit record will expire, causing a reset to the default begin-offset.

About heartbeats: heartbeats are sent in a dedicated socket without being blocked by the message handler processes. (The poll blocking heartbeats issue is quite common in early version Java client, but in newer version, they changed to a do heartbeats in a dedicated thread.)

However during the rebalance, coordinator has to wait for assignments revoked, this wait may cause rebalance timeout.

For your ref, here is a simplified case I encountered before:

Let’s call the large message set which blocks the handler for too long (or indefinitely) “poison pill”, and to make it easier to reason about, let’s assume there is only one such poison pill in one of partitions.

Initially the poison pill causes one member to hang during rebalance, then others will eventually reach balance without the hanging member, the partition which has the poison pill gets assigned to another member, causing the other to get poisoned.

If say, it takes 10 minutes to process the poison pill, and, if during this 10 minutes, all members are poisoned (due to rebalance triggered by whatever reason), then none of the consumers will be able to commit any offset.

if all members blocked for long enough, the commit record will expire, then all of a sudden, it can process again because the begin-offset has been reset.

Not sure if this is what happened in your case though.

The easy way out: if it takes longer time to process one message set than session timeout, it makes sense to increase both session timeout and rebalance timeout in consumer group config. (Changing default values of Kafka config will affect all other clients).

However increasing timeout implies longer time to reach balance, each time when a client (re)joins the group, it’ll have to wait for the slowest to revoke the assignments and rejoin the group.

A better solution is to implement async message handlers.

[matreyes]

@zmstone as you may expect, it was my mistake. I had 2 different callback modules, one for staging / dev and one for production.
In production I was responding {:ok, :ack, state} and in dev {:ok, :commit, state}, so dev commited and updated the consumer group, production didn't.
I've added the :ack when I was trying to solve the initial problem. So It wasn't the problem, the cause was in the monitoring system as you pointed out earlier.

Again, I'm very very grateful with your help and the work you are doing here. Without brod, I wouldn't be able to use elixir in my projects and the beam would be outdated in enterprise architectures. Github allowed me to sponsor you only with an espresso at Espresso House (lattes are too expensive in Stockholm). Enjoy!

Matías

[zmstone]

Great to hear that you finally have this resolved.
Thank you for your sponsor ;)