Skip to content

Commit f610995

Browse files
author
ssekaran
committed
Fixing file descriptor leak after file streamed in Sidecar Client
1 parent f0efd26 commit f610995

2 files changed

Lines changed: 185 additions & 10 deletions

File tree

‎analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/VertxHttpClient.java‎

Lines changed: 28 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,7 @@
5757
import org.apache.cassandra.sidecar.common.request.Request;
5858
import org.apache.cassandra.sidecar.common.request.UploadableRequest;
5959

60+
import static java.lang.String.valueOf;
6061
import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.AUTH_ROLE;
6162
import static org.apache.cassandra.sidecar.common.utils.StringUtils.isNullOrEmpty;
6263

@@ -165,11 +166,7 @@ protected CompletableFuture<HttpResponse> executeUploadFileInternal(SidecarInsta
165166
Promise<HttpResponse> promise = Promise.promise();
166167
// open the local file
167168
openFileForRead(vertx.fileSystem(), filename)
168-
.compose(pair -> vertxRequest.ssl(config.ssl())
169-
.putHeader(HttpHeaderNames.CONTENT_LENGTH.toString(),
170-
String.valueOf(pair.getKey()))
171-
.sendStream(pair.getValue()
172-
.setReadBufferSize(config.sendReadBufferSize())))
169+
.compose(pair -> sendFileStream(vertxRequest, pair, filename))
173170
.onFailure(promise::fail)
174171
.onSuccess(response -> {
175172
byte[] raw = response.body() != null ? response.body().getBytes() : null;
@@ -184,6 +181,32 @@ protected CompletableFuture<HttpResponse> executeUploadFileInternal(SidecarInsta
184181
return promise.future().toCompletionStage().toCompletableFuture();
185182
}
186183

184+
/**
185+
* Sends the file stream via HTTP request.
186+
*
187+
* @param vertxRequest the HTTP request to send the file stream with
188+
* @param pair a pair containing file size and the AsyncFile handle
189+
* @param filename the name of the file being uploaded (for logging purposes)
190+
* @return a Future that completes when the file has been sent
191+
*/
192+
protected Future<io.vertx.ext.web.client.HttpResponse<Buffer>> sendFileStream(
193+
HttpRequest<Buffer> vertxRequest,
194+
AbstractMap.SimpleEntry<Long, AsyncFile> pair,
195+
String filename)
196+
{
197+
AsyncFile asyncFile = pair.getValue();
198+
return vertxRequest.ssl(config.ssl())
199+
.putHeader(HttpHeaderNames.CONTENT_LENGTH.toString(),
200+
valueOf(pair.getKey()))
201+
.sendStream(pair.getValue()
202+
.setReadBufferSize(config.sendReadBufferSize()))
203+
.onComplete(ar -> {
204+
asyncFile.close().onFailure(err ->
205+
LOGGER.warn("Failed to close file after upload: filename='{}'", filename, err)
206+
);
207+
});
208+
}
209+
187210
/**
188211
* {@inheritDoc}
189212
*/

‎analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/VertxHttpClientTest.java‎

Lines changed: 157 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -18,16 +18,38 @@
1818

1919
package org.apache.cassandra.sidecar.client;
2020

21+
import java.io.IOException;
22+
import java.io.InputStream;
23+
import java.nio.file.Path;
24+
import java.nio.file.StandardCopyOption;
25+
import java.util.AbstractMap.SimpleEntry;
26+
import java.util.ArrayList;
27+
import java.util.List;
28+
import java.util.concurrent.TimeUnit;
29+
2130
import org.junit.jupiter.api.AfterAll;
31+
import org.junit.jupiter.api.AfterEach;
2232
import org.junit.jupiter.api.BeforeAll;
33+
import org.junit.jupiter.api.BeforeEach;
2334
import org.junit.jupiter.api.Test;
35+
import org.junit.jupiter.api.io.TempDir;
2436

37+
import io.vertx.core.Future;
2538
import io.vertx.core.Vertx;
2639
import io.vertx.core.buffer.Buffer;
40+
import io.vertx.core.file.AsyncFile;
2741
import io.vertx.ext.web.client.HttpRequest;
42+
import io.vertx.ext.web.client.HttpResponse;
43+
import okhttp3.mockwebserver.MockResponse;
44+
import okhttp3.mockwebserver.MockWebServer;
45+
import org.apache.cassandra.sidecar.client.request.RequestExecutorTest;
2846

47+
import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
48+
import static io.netty.handler.codec.http.HttpResponseStatus.OK;
49+
import static java.nio.file.Files.copy;
2950
import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.AUTH_ROLE;
3051
import static org.assertj.core.api.Assertions.assertThat;
52+
import static org.assertj.core.api.Assertions.assertThatThrownBy;
3153
import static org.mockito.Mockito.mock;
3254
import static org.mockito.Mockito.when;
3355

@@ -37,17 +59,35 @@
3759
public class VertxHttpClientTest
3860
{
3961
private static Vertx vertx;
62+
private MockWebServer mockServer;
63+
private HttpClientConfig config;
64+
private SidecarInstanceImpl sidecarInstance;
4065

41-
@BeforeAll
42-
public static void setUp()
66+
@BeforeEach
67+
public void setUp() throws IOException
4368
{
4469
vertx = Vertx.vertx();
70+
mockServer = new MockWebServer();
71+
mockServer.start();
72+
73+
config = new HttpClientConfig.Builder<>()
74+
.ssl(false)
75+
.timeoutMillis(30000)
76+
.build();
77+
sidecarInstance = RequestExecutorTest.newSidecarInstance(mockServer);
4578
}
4679

47-
@AfterAll
48-
public static void tearDown()
80+
@AfterEach
81+
public void tearDown() throws IOException
4982
{
50-
vertx.close();
83+
if (mockServer != null)
84+
{
85+
mockServer.shutdown();
86+
}
87+
if (vertx != null)
88+
{
89+
vertx.close();
90+
}
5191
}
5292

5393
@Test
@@ -74,4 +114,116 @@ private HttpClientConfig.Builder<?> httpClientConfigBuilder()
74114
.timeoutMillis(100)
75115
.idleTimeoutMillis(100);
76116
}
117+
118+
@Test
119+
void testUploadSSTableClosesFile(@TempDir Path tempDirectory) throws Exception
120+
{
121+
runTestScenario(tempDirectory,
122+
new MockResponse().setResponseCode(OK.code()),
123+
new ExposeAsyncFileVertxHttpClient(vertx, config));
124+
}
125+
126+
@Test
127+
void testUploadClosesFileOnUploadFailure(@TempDir Path tempDirectory) throws Exception
128+
{
129+
runTestScenario(tempDirectory,
130+
new MockResponse().setResponseCode(INTERNAL_SERVER_ERROR.code()),
131+
new ExposeAsyncFileVertxHttpClient(vertx, config));
132+
}
133+
134+
@Test
135+
void testMultipleUploadClosesAllFiles(@TempDir Path tempDirectory) throws Exception
136+
{
137+
mockServer.enqueue(new MockResponse().setResponseCode(OK.code()));
138+
mockServer.enqueue(new MockResponse().setResponseCode(OK.code()));
139+
mockServer.enqueue(new MockResponse().setResponseCode(OK.code()));
140+
141+
Path fileToUpload = prepareFile(tempDirectory);
142+
143+
ExposeAsyncFileVertxHttpClient httpClient = new ExposeAsyncFileVertxHttpClient(vertx, config);
144+
145+
// Upload the same file 3 times (simulating multiple file uploads)
146+
for (int i = 0; i < 3; i++)
147+
{
148+
HttpRequest<Buffer> vertxRequest = httpClient.webClient.put(mockServer.getPort(),
149+
mockServer.getHostName(),
150+
"/upload/test" + i);
151+
httpClient.executeUploadFileInternal(sidecarInstance, vertxRequest, fileToUpload.toString())
152+
.get(30, TimeUnit.SECONDS);
153+
}
154+
155+
assertThat(mockServer.getRequestCount()).isEqualTo(3);
156+
assertThat(httpClient.capturedFiles).hasSize(3);
157+
158+
// Give async file close operations time to complete
159+
Thread.sleep(100);
160+
161+
// Verify all the files are closed by attempting to call .end() which should throw IllegalStateException
162+
for (AsyncFile file : httpClient.capturedFiles)
163+
{
164+
assertThatThrownBy(file::end)
165+
.isInstanceOf(IllegalStateException.class)
166+
.hasMessageContaining("File handle is closed" );
167+
}
168+
}
169+
170+
private void runTestScenario(Path tempDirectory,
171+
MockResponse mockResponse,
172+
ExposeAsyncFileVertxHttpClient httpClient) throws Exception
173+
{
174+
mockServer.enqueue(mockResponse);
175+
176+
Path fileToUpload = prepareFile(tempDirectory);
177+
HttpRequest<Buffer> vertxRequest = httpClient.webClient.put(mockServer.getPort(),
178+
mockServer.getHostName(),
179+
"/upload/test" );
180+
181+
httpClient.executeUploadFileInternal(sidecarInstance, vertxRequest, fileToUpload.toString())
182+
.get(30, TimeUnit.SECONDS);
183+
184+
assertThat(mockServer.getRequestCount()).isEqualTo(1);
185+
186+
// Give async file close operation time to complete
187+
Thread.sleep(100);
188+
189+
// Verify file is closed by attempting to call .end() which should throw IllegalStateException
190+
assertThat(httpClient.capturedFiles).hasSize(1);
191+
assertThatThrownBy(() -> httpClient.capturedFiles.get(0).end())
192+
.isInstanceOf(IllegalStateException.class)
193+
.hasMessageContaining("File handle is closed" );
194+
}
195+
196+
/**
197+
* Class that extends from {@link VertxHttpClient} for testing purposes and holds a reference to the
198+
* {@link AsyncFile} to ensure that the file has been closed.
199+
*/
200+
static class ExposeAsyncFileVertxHttpClient extends VertxHttpClient
201+
{
202+
List<AsyncFile> capturedFiles = new ArrayList<>();
203+
204+
ExposeAsyncFileVertxHttpClient(Vertx vertx, HttpClientConfig config)
205+
{
206+
super(vertx, config);
207+
}
208+
209+
@Override
210+
protected Future<HttpResponse<Buffer>> sendFileStream(HttpRequest<Buffer> vertxRequest,
211+
SimpleEntry<Long, AsyncFile> pair,
212+
String filename)
213+
{
214+
capturedFiles.add(pair.getValue());
215+
return super.sendFileStream(vertxRequest, pair, filename);
216+
}
217+
}
218+
219+
private Path prepareFile(Path tempDirectory) throws IOException
220+
{
221+
Path fileToUpload = tempDirectory.resolve("nb-1-big-TOC.txt" );
222+
try (InputStream inputStream = getClass().getClassLoader().getResourceAsStream("sstables/nb-1-big-TOC.txt" ))
223+
{
224+
assertThat(inputStream).isNotNull();
225+
copy(inputStream, fileToUpload, StandardCopyOption.REPLACE_EXISTING);
226+
}
227+
return fileToUpload;
228+
}
77229
}

0 commit comments

Comments
 (0)