keith-turner commented on a change in pull request #369: [ACCUMULO-4787] Close
input stream in AccumuloReplicaSystem
URL: https://github.com/apache/accumulo/pull/369#discussion_r167342746
##########
File path:
server/tserver/src/main/java/org/apache/accumulo/tserver/replication/AccumuloReplicaSystem.java
##########
@@ -321,60 +321,67 @@ public String execute(ReplicationCoordinator.Client
client) throws Exception {
protected Status replicateRFiles(ClientContext peerContext, final
HostAndPort peerTserver, final ReplicationTarget target, final Path p, final
Status status,
final long sizeLimit, final String remoteTableId, final TCredentials
tcreds, final ReplicaSystemHelper helper, long timeout) throws
TTransportException,
AccumuloException, AccumuloSecurityException {
- DataInputStream input;
- try {
- input = getRFileInputStream(p);
- } catch (IOException e) {
- log.error("Could not create input stream from RFile, will retry", e);
- return status;
- }
-
- Status lastStatus = status, currentStatus = status;
- while (true) {
- // Read and send a batch of mutations
- ReplicationStats replResult =
ReplicationClient.executeServicerWithReturn(peerContext, peerTserver, new
RFileClientExecReturn(target, input, p,
- currentStatus, sizeLimit, remoteTableId, tcreds), timeout);
+ try (final DataInputStream input = getRFileInputStream(p)) {
+ Status lastStatus = status, currentStatus = status;
+ while (true) {
+ // Read and send a batch of mutations
+ ReplicationStats replResult =
ReplicationClient.executeServicerWithReturn(peerContext, peerTserver, new
RFileClientExecReturn(target, input, p,
+ currentStatus, sizeLimit, remoteTableId, tcreds), timeout);
- // Catch the overflow
- long newBegin = currentStatus.getBegin() + replResult.entriesConsumed;
- if (newBegin < 0) {
- newBegin = Long.MAX_VALUE;
- }
+ // Catch the overflow
+ long newBegin = currentStatus.getBegin() + replResult.entriesConsumed;
+ if (newBegin < 0) {
+ newBegin = Long.MAX_VALUE;
+ }
- currentStatus =
Status.newBuilder(currentStatus).setBegin(newBegin).build();
+ currentStatus =
Status.newBuilder(currentStatus).setBegin(newBegin).build();
- log.debug("Sent batch for replication of {} to {}, with new Status {}",
p, target, ProtobufUtil.toString(currentStatus));
+ log.debug("Sent batch for replication of {} to {}, with new Status
{}", p, target, ProtobufUtil.toString(currentStatus));
- // If we got a different status
- if (!currentStatus.equals(lastStatus)) {
- // If we don't have any more work, just quit
- if (!StatusUtil.isWorkRequired(currentStatus)) {
- return currentStatus;
+ // If we got a different status
+ if (!currentStatus.equals(lastStatus)) {
+ // If we don't have any more work, just quit
+ if (!StatusUtil.isWorkRequired(currentStatus)) {
+ return currentStatus;
+ } else {
+ // Otherwise, let it loop and replicate some more data
+ lastStatus = currentStatus;
+ }
} else {
- // Otherwise, let it loop and replicate some more data
- lastStatus = currentStatus;
- }
- } else {
- log.debug("Did not replicate any new data for {} to {}, (state was
{})", p, target, ProtobufUtil.toString(lastStatus));
+ log.debug("Did not replicate any new data for {} to {}, (state was
{})", p, target, ProtobufUtil.toString(lastStatus));
- // otherwise, we didn't actually replicate (likely because there was
error sending the data)
- // we can just not record any updates, and it will be picked up again
by the work assigner
- return status;
+ // otherwise, we didn't actually replicate (likely because there was
error sending the data)
+ // we can just not record any updates, and it will be picked up
again by the work assigner
+ return status;
+ }
}
+ } catch (IOException e) {
+ log.error("Could not create input stream from RFile, will retry", e);
+ return status;
}
}
protected Status replicateLogs(ClientContext peerContext, final HostAndPort
peerTserver, final ReplicationTarget target, final Path p, final Status status,
final long sizeLimit, final String remoteTableId, final TCredentials
tcreds, final ReplicaSystemHelper helper, final UserGroupInformation
accumuloUgi,
long timeout) throws TTransportException, AccumuloException,
AccumuloSecurityException {
-
log.debug("Replication WAL to peer tserver");
final Set<Integer> tids;
- final DataInputStream input;
- Span span = Trace.start("Read WAL header");
- span.data("file", p.toString());
- try {
- input = getWalStream(p);
+ try (final DataInputStream input = getWalStream(p)) {
Review comment:
The call to `getWalStream()` calls `DfsLogger.readHeaderAndReturnStream()`
which calls `fs.open()`. If there is IOException in
`readHeaderAndReturnStream()` or in `getWalStream()` then is seems like the
FSDataInputStream will not be closed.
Not sure what the best way to handle this is, but one possibility I thought
of is to refactor the code to call `fs.open()` in this method and then pass
that down. Maybe something like
```java
try(FSDataInputStream fsdi = fs.open(path)){
DataInputStream input = getWalStream(fsdi);
}
```
or could do the following, I think java will close things in the reverse
order from opened.
```java
try(FSDataInputStream fsdi = fs.open(path); DataInputStream input =
getWalStream(fsdi)){
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
