Skip to content

Better disconnection handling #21

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

Draft
wants to merge 1 commit into
base: master
Choose a base branch
from

Conversation

SuperCuber
Copy link

Goals:

  • Automatic detection of unrecoverable disconnection from broker
  • Convenient way for the application to query the healthiness of the consumer/producer

Maybe: Automatic recovery from the error

(if (seq records)
(run! put! records)
; Throws exception if not connected
(.listTopics consumer (Duration/ofMillis 5000)))
Copy link
Author

Choose a reason for hiding this comment

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

add one more timeout field to the config perhaps?

@@ -125,7 +126,10 @@

(loop []
(when-let [records (poll!)]
(run! put! records)
(if (seq records)
Copy link
Author

Choose a reason for hiding this comment

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

this means an additional listTopics network call every time an empty poll is received. Might have performance/load implications? I doubt it, since it only happens when 0 lag and the call should be pretty lightweight.

@@ -211,6 +215,9 @@
(defn done-chan [state]
(:ketu.source/consumer-thread state))

(defn closed? [consumer]
(-> consumer :ketu.source/out-chan async-protocols/closed?))
Copy link
Author

Choose a reason for hiding this comment

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

can also use done-chan

(run! put! records)
(if (seq records)
(run! put! records)
; Throws exception if not connected
Copy link
Author

Choose a reason for hiding this comment

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

do we want automatic recovery? or does it belong in the application / wrapper library?

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.

1 participant