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
Wait till node is part of cluster state for join process #6480
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -39,8 +39,14 @@ | |
*/ | ||
public class MembershipAction extends AbstractComponent { | ||
|
||
public static interface JoinCallback { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. at some point we need to find all classes that have this pattern and make them a generic class There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. aye, this would be nice |
||
void onSuccess(ClusterState state); | ||
|
||
void onFailure(Throwable t); | ||
} | ||
|
||
public static interface MembershipListener { | ||
ClusterState onJoin(DiscoveryNode node); | ||
void onJoin(DiscoveryNode node, JoinCallback callback); | ||
|
||
void onLeave(DiscoveryNode node); | ||
} | ||
|
@@ -160,13 +166,30 @@ public JoinRequest newInstance() { | |
} | ||
|
||
@Override | ||
public void messageReceived(JoinRequest request, TransportChannel channel) throws Exception { | ||
ClusterState clusterState = listener.onJoin(request.node); | ||
if (request.withClusterState) { | ||
channel.sendResponse(new JoinResponse(clusterState)); | ||
} else { | ||
channel.sendResponse(TransportResponse.Empty.INSTANCE); | ||
} | ||
public void messageReceived(final JoinRequest request, final TransportChannel channel) throws Exception { | ||
listener.onJoin(request.node, new JoinCallback() { | ||
@Override | ||
public void onSuccess(ClusterState state) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this error handling should happen upstream no? I mean we should call onFailure if onSuccess fails? I mean in theory we should have an abstract baseclass that handles that :) There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. yea, I didn't want to complicate the code to be honest to do so, would be nice to have a Callback base class that does that as you mentioned before, but I suggest in a different change |
||
try { | ||
if (request.withClusterState) { | ||
channel.sendResponse(new JoinResponse(state)); | ||
} else { | ||
channel.sendResponse(TransportResponse.Empty.INSTANCE); | ||
} | ||
} catch (Throwable t) { | ||
onFailure(t); | ||
} | ||
} | ||
|
||
@Override | ||
public void onFailure(Throwable t) { | ||
try { | ||
channel.sendResponse(t); | ||
} catch (Throwable e) { | ||
logger.warn("failed to send back failure on join request", e); | ||
} | ||
} | ||
}); | ||
} | ||
|
||
@Override | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
can this have some note when this was changed?