Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Defer source.finish() #157

Conversation

blindspotbounty
Copy link
Collaborator

Flush may throw exception thus source will not be finished.
Exception can be e.g. timeout (due to network outage):

Caught exception KafkaError.rdKafkaError: Local: Timed out Kafka/RDKafkaClient.swift...

@blindspotbounty blindspotbounty enabled auto-merge (squash) February 13, 2024 09:55
@blindspotbounty blindspotbounty enabled auto-merge (squash) February 13, 2024 16:13
Copy link
Collaborator

@mr-swifter mr-swifter left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Test

@mr-swifter mr-swifter self-requested a review February 14, 2024 07:35
@blindspotbounty blindspotbounty merged commit 20faefb into swift-server:main Feb 14, 2024
@blindspotbounty blindspotbounty deleted the producer-event-sequence-may-hang-on-stop branch March 4, 2024 10:43
blindspotbounty added a commit to ordo-one/swift-kafka-client that referenced this pull request Aug 6, 2024
* Feature: expose librdkafka statistics as swift metrics (swift-server#92)

* introduce statistics for producer

* add statistics to new consumer with events

* fix some artefacts

* adjust to KeyRefreshAttempts

* draft: statistics with metrics

* make structures internal

* Update Sources/Kafka/Configuration/KafkaConfiguration+Metrics.swift

Co-authored-by: Felix Schlegel <[email protected]>

* Update Sources/Kafka/Configuration/KafkaConsumerConfiguration.swift

Co-authored-by: Felix Schlegel <[email protected]>

* Update Sources/Kafka/Configuration/KafkaConfiguration+Metrics.swift

Co-authored-by: Felix Schlegel <[email protected]>

* Update Sources/Kafka/Configuration/KafkaConfiguration+Metrics.swift

Co-authored-by: Felix Schlegel <[email protected]>

* address review comments

* formatting

* map gauges in one place

* move json mode as rd kafka statistics, misc renaming + docc

* address review comments

* remove import Metrics

* divide producer/consumer configuration

* apply swiftformat

* fix code after conflicts

* fix formatting

---------

Co-authored-by: Felix Schlegel <[email protected]>

* Add benchmark infratructure without actual tests (swift-server#146)

* add benchmark infratructure without actual test

* apply swiftformat

* fix header in sh file

* use new async seq methods

* Update to latest librdkafka & add a define for RAND_priv_bytes (swift-server#148)

Co-authored-by: Franz Busch <[email protected]>

* exit from consumer batch loop when no more messages left (swift-server#153)

* Lower requirements for consumer state machine (swift-server#154)

* lower requirements for kafka consumer

* add twin test for kafka producer

* defer source.finish (swift-server#157)

* Add two consumer benchmark (swift-server#149)

* benchmark for consumer

* attempty to speedup benchmarks

* check CI works for one test

* enable one more test

* try to lower poll interval

* adjust max duration of test

* remain only manual commit test

* check if commit is the reason for test delays

* try all with schedule commit

* revert max test time to 5 seconds

* dockerfiles

* test set threasholds

* create dummy thresholds from ci results

* disable benchmark in CI

* add header

* add stable metrics

* update thresholds to stable metrics only

* try use '1' instead of 'true'

* adjust thresholds to CI results (as temporary measure)

* set 20% threshold..

* move arc to unstable metrics

* try use 'true' in quotes for CI

* try reduce number of messages for more reliable results

* try upgrade bench

* disable benchmark in CI

* Update librdkafka for BoringSSL (swift-server#162)

* chore(patch): [sc-8379] use returned error (swift-server#163)

* [producer message] Allow optional key for initializer (swift-server#164)

Co-authored-by: Harish Yerra <[email protected]>

* Allow groupID to be specified when assigning partition (swift-server#161)

* Allow groupID to be specified when assigning partition

Motivation:

A Consumer Group can provide a lot of benefits even if the
dynamic loadbalancing features are not used.

Modifications:

Allow for an optional GroupID when creating a partition
consumer.

Result:

Consumer Groups can now be used when manual assignment is
used.

* fix format

---------

Co-authored-by: Ómar Kjartan Yasin <[email protected]>
Co-authored-by: blindspotbounty <[email protected]>
Co-authored-by: Franz Busch <[email protected]>

* Wrap rd_kafka_consumer_poll into iterator (use librdkafka embedded backpressure) (swift-server#158)

* remove message sequence

* test consumer with implicit rebalance

* misc + format

* remove artefact

* don't check a lot of messages

* fix typo

* slow down first consumer to lower message to fit CI timeout

* remove helpers

* use exact benchmark version to avoid missing thresholds error (as no thresholds so far)

* add deprecated marks for backpressure, change comment for future dev

* address comments

---------

Co-authored-by: Felix Schlegel <[email protected]>
Co-authored-by: Axel Andersson <[email protected]>
Co-authored-by: Franz Busch <[email protected]>
Co-authored-by: Samuel M <[email protected]>
Co-authored-by: Harish Yerra <[email protected]>
Co-authored-by: Harish Yerra <[email protected]>
Co-authored-by: Omar Yasin <[email protected]>
Co-authored-by: Ómar Kjartan Yasin <[email protected]>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

2 participants