Skip to content

HDDS-2347. XCeiverClientGrpc's parallel use leads to NPE #81

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.

Sign up for GitHub

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

Merged
lokeshj1703 merged 1 commit into from
Oct 30, 2019
Merged

HDDS-2347. XCeiverClientGrpc's parallel use leads to NPE #81

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 23 additions & 15 deletions hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,8 +146,11 @@ public void connect(String encodedToken) throws Exception {
connectToDatanode(dn, encodedToken);
}

private void connectToDatanode(DatanodeDetails dn, String encodedToken)
throws IOException {
private synchronized void connectToDatanode(DatanodeDetails dn,
String encodedToken) throws IOException {
if (isConnected(dn)){
return;
}
// read port from the data node, on failure use default configured
// port.
int port = dn.getPort(DatanodeDetails.Port.Name.STANDALONE).getValue();
Expand Down Expand Up @@ -202,7 +205,7 @@ private boolean isConnected(ManagedChannel channel) {
}

@Override
public void close() {
public synchronized void close() {
closed = true;
for (ManagedChannel channel : channels.values()) {
channel.shutdownNow();
Expand Down Expand Up @@ -368,19 +371,9 @@ public XceiverClientReply sendCommandAsync(

private XceiverClientReply sendCommandAsync(
ContainerCommandRequestProto request, DatanodeDetails dn)
throws IOException, ExecutionException, InterruptedException {
if (closed) {
throw new IOException("This channel is not connected.");
}

throws IOException, InterruptedException {
checkOpen(dn, request.getEncodedToken());
UUID dnId = dn.getUuid();
ManagedChannel channel = channels.get(dnId);
// If the channel doesn't exist for this specific datanode or the channel
// is closed, just reconnect
String token = request.getEncodedToken();
if (!isConnected(channel)) {
reconnect(dn, token);
}
if (LOG.isDebugEnabled()) {
LOG.debug("Send command {} to datanode {}",
request.getCmdType().toString(), dn.getNetworkFullPath());
Expand Down Expand Up @@ -427,6 +420,21 @@ public void onCompleted() {
return new XceiverClientReply(replyFuture);
}

private synchronized void checkOpen(DatanodeDetails dn, String encodedToken)
throws IOException{
if (closed) {
throw new IOException("This channel is not connected.");
}

ManagedChannel channel = channels.get(dn.getUuid());
// If the channel doesn't exist for this specific datanode or the channel
// is closed, just reconnect
if (!isConnected(channel)) {
reconnect(dn, encodedToken);
}

}

private void reconnect(DatanodeDetails dn, String encodedToken)
throws IOException {
ManagedChannel channel;
Expand Down
2 changes: 1 addition & 1 deletion hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,7 +235,6 @@ public XceiverClientSpi call() throws Exception {
case RATIS:
client = XceiverClientRatis.newXceiverClientRatis(pipeline, conf,
caCert);
client.connect();
break;
case STAND_ALONE:
client = new XceiverClientGrpc(pipeline, conf, caCert);
Expand All @@ -244,6 +243,7 @@ public XceiverClientSpi call() throws Exception {
default:
throw new IOException("not implemented" + pipeline.getType());
}
client.connect();
return client;
}
});
Expand Down