Remove raise and handle reconnection #1
Closed
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
IanVaughan
force-pushed
the
remove-raise-and-handle-reconnection
branch
2 times, most recently
from
abba1a7 to
f7bf4e7
Compare
luizvarela
reviewed
Aug 16, 2018
IanVaughan
force-pushed
the
remove-raise-and-handle-reconnection
branch
from
f7bf4e7 to
1cee8bf
Compare
qqdanielqq
reviewed
Aug 17, 2018
|
rebased and sorted commits, just need to confirm order and double check last "stash" commit. |
IanVaughan
force-pushed
the
remove-raise-and-handle-reconnection
branch
2 times, most recently
from
fe91475 to
d71163f
Compare
added 7 commits
August 17, 2018 11:06
By attempting to connect within the init/1 callback, we have no room to reconnect, as the init raises, causing the KafkaEx.ConsumerGroup.init/1 to fail when attempting to supervise a worker, this then crashes the application. Save the worker_opts for a new GenServer callback to use No need to set member_id to nil within init/1 as it will already be nil Handle create_worker and other phases within callbacks. Each phase once successful sends a message to itself to move onto the next phase, allowing attempts to be retried.
This removes the connection from the init/1 callback,
so it can retry.
As this module is supervised by the main OTP app,
if it returns stop too many times it will stop the app.
eg {:stop, reason} can not be used.
Could consider :ignore, but returns `{:ok, ?}`
which means start_worker cannot pattern match
Returns the value of the worker Server module
The same could be said for the following :timeout function that calls consume, but as this matches first after connection once this completes it is known to be ready as its set the current_offset
IanVaughan
force-pushed
the
remove-raise-and-handle-reconnection
branch
from
d71163f to
ec847d6
Compare
added 11 commits
August 17, 2018 11:33
But allow happy path to use faster @startup_delay
So the socket connection can be obtained if the broker is down during init but then comes back
During reconnection the default can be attempted
14:52:19.028 [error] GenServer #PID<0.196.0> terminating
** (FunctionClauseError) no function clause matching in KafkaEx.NetworkClient.send_sync_request/3
(kafka_ex) lib/kafka_ex/network_client.ex:35: KafkaEx.NetworkClient.send_sync_request(nil, <<0, 11, 0, 0, 0, 0, 0, 8, 0, 8, 107, 97, 102, 107, 97, 95, 101, 120, 0, 18, 116, 114, 97, 99, 107, 105, 110, 103, 95, 108, 111, 99, 97, 116, 105, 111, 110, 115, 0, 0, 117, 48, 0, 0, 0, 8, 99, 111, 110, 115, ...>>, 35000)
(kafka_ex) lib/kafka_ex/server_0_p_9_p_0.ex:189: KafkaEx.Server0P9P0.consumer_group_sync_request/4
(kafka_ex) lib/kafka_ex/server_0_p_9_p_0.ex:116: KafkaEx.Server0P9P0.kafka_server_join_group/3
otherwise it updates the state with invalid metadata, which then attempts to read etc, and crashes for the wrong reasons, even though we have lost contact with the broker and want to exit anyway.
This ensures that stopped genserver call does not blow this process up as well
14:57:08.753 [error] GenServer #PID<0.182.0> terminating
** (stop) "Unable to fetch metadata from any brokers. Timeout is 3000."
Last message: {:metadata, ""}
|
This is really useful and exactly what I'm looking for! Confused why this PR is not on the main kafka_ex repo? ( https://github.com/kafkaex/kafka_ex ) Is that on purpose? |
|
Closing as its been too long and I do not have the time to finish now. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Initial POC of reconnection handling, attempting to solve kafkaex#298
Main change is to remove the
raiseif the broker cannot be found, but in doing that means theinit/1has to return something, and if the host is not found it could return{:error, reason}, but that causes theKafkaEx.ConsumerGroup.Managersupervisor that calls it issues, should it then error as well?Instead, I split the elixir boot sequence from the kafka connection, as the elixir startup should be successful regardless of outside influences and then handle connecting.
TODO