Close InputStream when receiving cluster state in PublishClusterStateAction
Not closing the InputStream will leak native memory as the DeflateCompressor/Inflater won't be closed.
package org.elasticsearch.discovery.zen;
import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.apache.lucene.util.IOUtils;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.Version;
import org.elasticsearch.cluster.ClusterChangedEvent;
@@ -375,11 +376,10 @@ public class PublishClusterStateAction extends AbstractComponent {
protected void handleIncomingClusterStateRequest(BytesTransportRequest request, TransportChannel channel) throws IOException {
Compressor compressor = CompressorFactory.compressor(request.bytes());
- StreamInput in;
+ StreamInput in = request.bytes().streamInput();
+ try {
if (compressor != null) {
- in = compressor.streamInput(request.bytes().streamInput());
- } else {
- in = request.bytes().streamInput();
+ in = compressor.streamInput(in);
}
in = new NamedWriteableAwareStreamInput(in, namedWriteableRegistry);
in.setVersion(request.version());
@@ -388,7 +388,8 @@ public class PublishClusterStateAction extends AbstractComponent {
// If true we received full cluster state - otherwise diffs
if (in.readBoolean()) {
incomingState = ClusterState.readFrom(in, clusterStateSupplier.get().nodes().getLocalNode());
- logger.debug("received full cluster state version [{}] with size [{}]", incomingState.version(), request.bytes().length());
+ logger.debug("received full cluster state version [{}] with size [{}]", incomingState.version(),
+ request.bytes().length());
} else if (lastSeenClusterState != null) {
Diff<ClusterState> diff = ClusterState.readDiffFrom(in, lastSeenClusterState.nodes().getLocalNode());
incomingState = diff.apply(lastSeenClusterState);
@@ -404,6 +405,9 @@ public class PublishClusterStateAction extends AbstractComponent {
pendingStatesQueue.addPending(incomingState);
lastSeenClusterState = incomingState;
}
+ } finally {
+ IOUtils.close(in);
+ }
channel.sendResponse(TransportResponse.Empty.INSTANCE);
}
