Add share group consumer

[orangecola]

Fixes #251

E2E Testing:

$ docker exec ls-test sh -c 'pkill -f logstash; rm -f /tmp/share_output.jsonl /tmp/ls.log; sleep 2' 2>&1
docker exec kafka-sg /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic final-test --partitions 3 --replication-factor 1 2>&1
docker exec kafka-sg /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config share.auto.offset.reset=earliest --entity-type groups --entity-name final-group 2>&1
for i in $(seq 1 50); do echo "final-$i"; done | docker exec -i kafka-sg /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic final-test 2>&1 | tail -1

cat > /tmp/final.conf <<'EOF'
input {
  kafka {
    bootstrap_servers => "kafka-sg:9092"
    topics => ["final-test"]
    group_id => "final-group"
    consumer_mode => "share_group"
    consumer_threads => 3
    client_id => "final-smoke"
    codec => "plain"
    poll_timeout_ms => 500
  }
}
output {
  file { path => "/tmp/share_output.jsonl" codec => line { format => "%{message}" } }
}
EOF
docker cp /tmp/final.conf ls-test:/tmp/final.conf
docker exec -d ls-test sh -c '/usr/share/logstash/bin/logstash -f /tmp/final.conf --path.data /tmp/ls-data-final > /tmp/ls.log 2>&1'
for i in 1 2 3 4 5 6 7 8; do
  sleep 5
  count=$(docker exec ls-test sh -c 'test -f /tmp/share_output.jsonl && wc -l < /tmp/share_output.jsonl || echo 0' 2>&1 | tr -d ' ')
  echo "t=${i}x5s: received=$count"
  if [ "$count" = "50" ]; then break; fi
done
echo "=== SUMMARY ==="
docker exec ls-test sh -c 'echo "total=$(wc -l < /tmp/share_output.jsonl)  unique=$(sort -u /tmp/share_output.jsonl | wc -l)  duplicates=$(sort /tmp/share_output.jsonl | uniq -d | wc -l)"' 2>&1
echo "=== error check ==="
docker exec ls-test sh -c 'grep -cE "\[ERROR\]" /tmp/ls.log' 2>&1
docker exec ls-test sh -c 'grep -E "\[ERROR\]" /tmp/ls.log | grep -v -E "licensereader|monitoring"' 2>&1 | head -5

Created topic final-test.
Completed updating config for group final-group.
t=1x5s: received=37
t=2x5s: received=37
t=3x5s: received=37
t=4x5s: received=87
t=5x5s: received=87
t=6x5s: received=87
t=7x5s: received=87
t=8x5s: received=87
=== SUMMARY ===
total=87  unique=85  duplicates=1
=== error check ===
3

[alexcams]

Thanks for this contribution @orangecola! The overall implementation looks nice, just left some nits worth to check.
Could you also update CHANGELOG.md file and bump version in gempspec?

[orangecola]

Addressed comments and updated version files.

Thanks for this contribution @orangecola! The overall implementation looks nice, just left some nits worth to check. Could you also update CHANGELOG.md file and bump version in gempspec?

[alexcams]

Thanks for addressing the comments @orangecola! I've been testing this feature manually and I can confirm the share group consumer works correctly — records are distributed across threads without specific thread-partition assignment.

While checking the docs, I've seen a recommendation for consumer_threads config that states:

Ideally you should have as many threads as the number of partitions for a perfect balance -- more threads than partitions means that some threads will be idle

in 3 places:

• http://github.com/logstash-plugins/logstash-integration-kafka/blob/main/docs/input-kafka.asciidoc?plain=1#L61
• https://github.com/logstash-plugins/logstash-integration-kafka/blob/main/docs/input-kafka.asciidoc?plain=1#L294
• https://github.com/logstash-plugins/logstash-integration-kafka/blob/main/lib/logstash/inputs/kafka.rb#L103

Would you mind updating this to be consistent with the new share_group consumer mode?
Thanks in advance!

[alexcams]

Requested changes in last review

[orangecola]

mb, didn't push changes.

[alexcams]

Thanks @orangecola!

[mashhurs]

I believe this feature will be 🚀. Thank you alot for the contribution.
The implementation details, docs and format look good to me as well (will review docs section with second round).

One critical concern is that when pipeline config contains the params which cannot be used with share_group, I would support raising a config error instead of silently ignoring. For example, the following config looks invalid to me:

input {
  kafka {
    enable_auto_commit => true
    consumer_mode => "share_group"
  }
}

[mashhurs]

not_allowed_configs_with_share_group =  original_params.keys & %w( group_protocol group_instance_id partition_assignment_strategy enable_auto_commit auto_commit_interval_ms)

if not_allowed_configs_with_share_group.size > 1
  error_message = "#{not_allowed_configs_with_share_group} options cannot be used together when `consumer_mode => 'share_group'`"
  raise LogStash::ConfigurationError, error_message
end

[mashhurs]

I will also pull, run on my local and review in detail but I would be really specific and not allow enable_auto_commit, heartbeat_interval_ms, etc... params if pipeline config contains. The original_params.keys provides the pipeline config param keys (I will comment below how it can be used).

[orangecola]

Combined schema_registry_url with other field checks.

[yaauie]

Overall, excellent work and a welcome contribution. Thank you!

I've left a few comments in-line, addressing:

• translating the kafka-ish docs to how the settings apply in the context of a logstash pipeline
• validating that share_group-specific config options aren't used in conjunction with the consumer_mode => consumer_group (an inversion of the check already added that ensures that consumer_group-specific settings aren't used with the new consumer_mode => share_group)

[cla-checker-service]

💚 CLA has been signed

[orangecola]

@yaauie / @alexcams updated with suggestions, bumping for review.