Skip to content

Commit ae393e6

Browse files
nodeceCopilot
authored and
priyanshu-ctds
committed
[improve][proxy] Add regression tests for package upload with 'Expect: 100-continue' (apache#25211)
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> (cherry picked from commit e8fedb1) (cherry picked from commit 0947639)
1 parent d6d5c6b commit ae393e6

5 files changed

Lines changed: 220 additions & 73 deletions

File tree

‎pulsar-package-management/core/src/test/java/org/apache/pulsar/packages/management/core/MockedPackagesStorage.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
*/
1919
package org.apache.pulsar.packages.management.core;
2020

21+
import java.io.ByteArrayOutputStream;
2122
import java.io.IOException;
2223
import java.io.InputStream;
2324
import java.io.OutputStream;
@@ -45,8 +46,13 @@ public CompletableFuture<Void> writeAsync(String path, InputStream inputStream)
4546
CompletableFuture<Void> future = new CompletableFuture<>();
4647
CompletableFuture.runAsync(() -> {
4748
try {
48-
byte[] bytes = new byte[inputStream.available()];
49-
inputStream.read(bytes);
49+
ByteArrayOutputStream baos = new ByteArrayOutputStream();
50+
byte[] buffer = new byte[8192];
51+
int read;
52+
while ((read = inputStream.read(buffer)) != -1) {
53+
baos.write(buffer, 0, read);
54+
}
55+
byte[] bytes = baos.toByteArray();
5056
storage.put(path, bytes);
5157
future.complete(null);
5258
} catch (IOException e) {

‎pulsar-package-management/core/src/test/java/org/apache/pulsar/packages/management/core/MockedPackagesStorageTest.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.pulsar.packages.management.core;
20+
21+
import static org.mockito.Mockito.mock;
22+
import static org.testng.Assert.assertEquals;
23+
import java.io.ByteArrayInputStream;
24+
import java.io.ByteArrayOutputStream;
25+
import lombok.Cleanup;
26+
import org.testng.annotations.Test;
27+
28+
public class MockedPackagesStorageTest {
29+
30+
@Test
31+
public void testWriteAndRead() throws Exception {
32+
PackagesStorageProvider provider = new MockedPackagesStorageProvider();
33+
PackagesStorage storage = provider.getStorage(mock(PackagesStorageConfiguration.class));
34+
storage.initialize();
35+
36+
// Test data
37+
byte[] testBytes = new byte[1 * 1024 * 1024];
38+
39+
// Write
40+
storage.writeAsync("test/path", new ByteArrayInputStream(testBytes)).get();
41+
42+
// Read
43+
@Cleanup
44+
ByteArrayOutputStream readBaos = new ByteArrayOutputStream();
45+
storage.readAsync("test/path", readBaos).get();
46+
47+
// Verify
48+
assertEquals(readBaos.toByteArray(), testBytes);
49+
50+
storage.closeAsync().get();
51+
}
52+
}

‎pulsar-proxy/pom.xml‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -227,6 +227,14 @@
227227
<type>test-jar</type>
228228
<scope>test</scope>
229229
</dependency>
230+
231+
<dependency>
232+
<groupId>${project.groupId}</groupId>
233+
<artifactId>pulsar-package-core</artifactId>
234+
<version>${project.version}</version>
235+
<type>test-jar</type>
236+
<scope>test</scope>
237+
</dependency>
230238
</dependencies>
231239
<build>
232240
<plugins>

‎pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java‎

Lines changed: 8 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -29,12 +29,10 @@
2929
import java.util.HashSet;
3030
import java.util.Iterator;
3131
import java.util.Set;
32-
import java.util.concurrent.Executor;
3332
import java.util.concurrent.Executors;
3433
import java.util.concurrent.ScheduledExecutorService;
3534
import java.util.concurrent.TimeUnit;
3635
import javax.net.ssl.SSLContext;
37-
import javax.servlet.ServletConfig;
3836
import javax.servlet.ServletException;
3937
import javax.servlet.http.HttpServletRequest;
4038
import javax.servlet.http.HttpServletResponse;
@@ -56,9 +54,7 @@
5654
import org.eclipse.jetty.client.http.HttpClientTransportOverHTTP;
5755
import org.eclipse.jetty.http.HttpHeader;
5856
import org.eclipse.jetty.proxy.ProxyServlet;
59-
import org.eclipse.jetty.util.HttpCookieStore;
6057
import org.eclipse.jetty.util.ssl.SslContextFactory;
61-
import org.eclipse.jetty.util.thread.QueuedThreadPool;
6258
import org.slf4j.Logger;
6359
import org.slf4j.LoggerFactory;
6460

@@ -118,75 +114,16 @@ class AdminProxyHandler extends ProxyServlet {
118114

119115
@Override
120116
protected HttpClient createHttpClient() throws ServletException {
121-
ServletConfig config = getServletConfig();
122-
123-
HttpClient client = newHttpClient();
124-
125-
client.setFollowRedirects(true);
126-
127-
// Must not store cookies, otherwise cookies of different clients will mix.
128-
client.setCookieStore(new HttpCookieStore.Empty());
129-
130-
Executor executor;
131-
String value = config.getInitParameter("maxThreads");
132-
if (value == null || "-".equals(value)) {
133-
executor = (Executor) getServletContext().getAttribute("org.eclipse.jetty.server.Executor");
134-
if (executor == null) {
135-
throw new IllegalStateException("No server executor for proxy");
136-
}
137-
} else {
138-
QueuedThreadPool qtp = new QueuedThreadPool(Integer.parseInt(value));
139-
String servletName = config.getServletName();
140-
int dot = servletName.lastIndexOf('.');
141-
if (dot >= 0) {
142-
servletName = servletName.substring(dot + 1);
143-
}
144-
qtp.setName(servletName);
145-
executor = qtp;
146-
}
147-
148-
client.setExecutor(executor);
149-
150-
value = config.getInitParameter("maxConnections");
151-
if (value == null) {
152-
value = "256";
153-
}
154-
client.setMaxConnectionsPerDestination(Integer.parseInt(value));
155-
156-
value = config.getInitParameter("idleTimeout");
157-
if (value == null) {
158-
value = "30000";
159-
}
160-
client.setIdleTimeout(Long.parseLong(value));
161-
162-
value = config.getInitParameter(INIT_PARAM_REQUEST_BUFFER_SIZE);
163-
if (value != null) {
164-
client.setRequestBufferSize(Integer.parseInt(value));
165-
}
166-
167-
value = config.getInitParameter("responseBufferSize");
168-
if (value != null){
169-
client.setResponseBufferSize(Integer.parseInt(value));
170-
}
171-
172-
try {
173-
client.start();
174-
175-
// Content must not be decoded, otherwise the client gets confused.
176-
// Allow encoded content, such as "Content-Encoding: gzip", to pass through without decoding it.
177-
client.getContentDecoderFactories().clear();
178-
179-
// Pass traffic to the client, only intercept what's necessary.
180-
ProtocolHandlers protocolHandlers = client.getProtocolHandlers();
181-
protocolHandlers.clear();
182-
protocolHandlers.put(new RedirectProtocolHandler(client));
183-
184-
return client;
185-
} catch (Exception x) {
186-
throw new ServletException(x);
187-
}
117+
HttpClient httpClient = super.createHttpClient();
118+
customizeHttpClient(httpClient);
119+
return httpClient;
188120
}
189121

122+
private void customizeHttpClient(HttpClient httpClient) {
123+
httpClient.setFollowRedirects(true);
124+
ProtocolHandlers protocolHandlers = httpClient.getProtocolHandlers();
125+
protocolHandlers.put(new RedirectProtocolHandler(httpClient));
126+
}
190127

191128
// This class allows the request body to be replayed, the default implementation
192129
// does not

‎pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyPackagesUploadTest.java‎

Lines changed: 144 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,144 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.pulsar.proxy.server;
20+
21+
import static com.google.common.net.HttpHeaders.EXPECT;
22+
import static org.assertj.core.api.Assertions.assertThat;
23+
import com.fasterxml.jackson.databind.ObjectMapper;
24+
import java.nio.file.Files;
25+
import java.nio.file.Path;
26+
import java.util.Optional;
27+
import lombok.Cleanup;
28+
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
29+
import org.apache.pulsar.broker.authentication.AuthenticationService;
30+
import org.apache.pulsar.client.admin.PulsarAdmin;
31+
import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
32+
import org.apache.pulsar.common.util.ObjectMapperFactory;
33+
import org.apache.pulsar.packages.management.core.MockedPackagesStorageProvider;
34+
import org.apache.pulsar.packages.management.core.common.PackageMetadata;
35+
import org.asynchttpclient.AsyncHttpClient;
36+
import org.asynchttpclient.DefaultAsyncHttpClient;
37+
import org.asynchttpclient.DefaultAsyncHttpClientConfig;
38+
import org.asynchttpclient.RequestBuilder;
39+
import org.asynchttpclient.Response;
40+
import org.asynchttpclient.request.body.multipart.FilePart;
41+
import org.asynchttpclient.request.body.multipart.StringPart;
42+
import org.eclipse.jetty.servlet.ServletHolder;
43+
import org.testng.annotations.AfterMethod;
44+
import org.testng.annotations.BeforeMethod;
45+
import org.testng.annotations.Test;
46+
47+
@Test(groups = "broker-admin")
48+
public class ProxyPackagesUploadTest extends MockedPulsarServiceBaseTest {
49+
50+
private static final int FILE_SIZE = 8 * 1024 * 1024; // 8 MB
51+
private static final ObjectMapper MAPPER = ObjectMapperFactory.create();
52+
private WebServer webServer;
53+
private PulsarAdmin proxyAdmin;
54+
55+
@BeforeMethod(alwaysRun = true)
56+
@Override
57+
protected void setup() throws Exception {
58+
conf.setEnablePackagesManagement(true);
59+
conf.setPackagesManagementStorageProvider(MockedPackagesStorageProvider.class.getName());
60+
super.internalSetup();
61+
62+
ProxyConfiguration proxyConfig = new ProxyConfiguration();
63+
proxyConfig.setServicePort(Optional.of(0));
64+
proxyConfig.setWebServicePort(Optional.of(0));
65+
proxyConfig.setBrokerWebServiceURL(brokerUrl.toString());
66+
67+
webServer = new WebServer(proxyConfig, new AuthenticationService(
68+
PulsarConfigurationLoader.convertFrom(proxyConfig, true)));
69+
webServer.addServlet("/", new ServletHolder(new AdminProxyHandler(proxyConfig, null, null)));
70+
webServer.start();
71+
72+
proxyAdmin = PulsarAdmin.builder()
73+
.serviceHttpUrl("http://localhost:" + webServer.getListenPortHTTP().get())
74+
.build();
75+
76+
admin.tenants().createTenant("public", createDefaultTenantInfo());
77+
admin.namespaces().createNamespace("public/default");
78+
}
79+
80+
@AfterMethod(alwaysRun = true)
81+
@Override
82+
protected void cleanup() throws Exception {
83+
if (proxyAdmin != null) {
84+
proxyAdmin.close();
85+
}
86+
if (webServer != null) {
87+
webServer.stop();
88+
}
89+
super.internalCleanup();
90+
}
91+
92+
@Test
93+
public void testUploadPackageThroughProxy() throws Exception {
94+
Path packageFile = Files.createTempFile("pkg-sdk", ".nar");
95+
packageFile.toFile().deleteOnExit();
96+
Files.write(packageFile, new byte[FILE_SIZE]);
97+
98+
String pkgName = "function://public/default/pkg-sdk@v1";
99+
PackageMetadata meta = PackageMetadata.builder().description("sdk-test").build();
100+
101+
proxyAdmin.packages().upload(meta, pkgName, packageFile.toString());
102+
103+
verifyDownload(pkgName, FILE_SIZE);
104+
}
105+
106+
@Test
107+
public void testUploadWithExpect100Continue() throws Exception {
108+
Path packageFile = Files.createTempFile("pkg-ahc", ".nar");
109+
packageFile.toFile().deleteOnExit();
110+
Files.write(packageFile, new byte[FILE_SIZE]);
111+
112+
String pkgName = "function://public/default/expect-test@v1";
113+
String uploadUrl = String.format("http://localhost:%d/admin/v3/packages/function/public/default/expect-test/v1",
114+
webServer.getListenPortHTTP().orElseThrow());
115+
116+
@Cleanup
117+
AsyncHttpClient client = new DefaultAsyncHttpClient(new DefaultAsyncHttpClientConfig.Builder().build());
118+
119+
Response response = client.executeRequest(new RequestBuilder("POST")
120+
.setUrl(uploadUrl)
121+
.addHeader(EXPECT, "100-continue")
122+
.addBodyPart(new FilePart("file", packageFile.toFile()))
123+
.addBodyPart(new StringPart("metadata", MAPPER.writeValueAsString(
124+
PackageMetadata.builder().description("ahc-test").build()), "application/json"))
125+
.build()).get();
126+
127+
assertThat(response.getStatusCode()).isEqualTo(204);
128+
129+
verifyDownload(pkgName, FILE_SIZE);
130+
}
131+
132+
private void verifyDownload(String packageName, int expectedSize) throws Exception {
133+
Path fromBroker = Files.createTempFile("from-broker", ".nar");
134+
fromBroker.toFile().deleteOnExit();
135+
admin.packages().download(packageName, fromBroker.toString());
136+
assertThat(Files.size(fromBroker)).isEqualTo(expectedSize);
137+
Files.deleteIfExists(fromBroker);
138+
139+
Path fromProxy = Files.createTempFile("from-proxy", ".nar");
140+
fromProxy.toFile().deleteOnExit();
141+
proxyAdmin.packages().download(packageName, fromProxy.toString());
142+
assertThat(Files.size(fromProxy)).isEqualTo(expectedSize);
143+
}
144+
}

0 commit comments

Comments
 (0)