Skip to content

Avoid reusing released APIServer response buffers #273

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
popduke merged 1 commit into from
Aug 12, 2026
Merged

Avoid reusing released APIServer response buffers #273

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
14 changes: 7 additions & 7 deletions bifromq-apiserver/src/main/java/org/apache/bifromq/apiserver/http/handler/KillHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@
import static org.apache.bifromq.apiserver.http.handler.utils.HeaderUtils.getHeader;

import com.google.common.base.Strings;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
Expand All @@ -47,6 +46,7 @@
import io.swagger.v3.oas.annotations.responses.ApiResponses;
import jakarta.ws.rs.DELETE;
import jakarta.ws.rs.Path;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
Expand All @@ -65,10 +65,8 @@ final class KillHandler extends TenantAwareHandler {
private static final String SERVER_REDIRECT_VALUE_TEMP_USE = "temp_use";
private static final Set<String> SERVER_REDIRECT_VALUES =
Set.of(SERVER_REDIRECT_VALUE_NO, SERVER_REDIRECT_VALUE_MOVE, SERVER_REDIRECT_VALUE_TEMP_USE);
private static final ByteBuf INVALID_SERVER_REDIRECT =
Unpooled.wrappedBuffer("Invalid server redirect value".getBytes());
private static final ByteBuf TOO_LONG_SERVER_REFERENCE =
Unpooled.wrappedBuffer("Server reference exceeds 65535 bytes".getBytes());
private static final String INVALID_SERVER_REDIRECT = "Invalid server redirect value";
private static final String TOO_LONG_SERVER_REFERENCE = "Server reference exceeds 65535 bytes";
private final ISessionDictClient sessionDictClient;

KillHandler(ISettingProvider settingProvider, ISessionDictClient sessionDictClient) {
Expand Down Expand Up @@ -119,11 +117,13 @@ public CompletableFuture<FullHttpResponse> handle(@Parameter(hidden = true) long
Map<String, String> clientMeta = getClientMeta(req);
if (serverRedirect != null && !SERVER_REDIRECT_VALUES.contains(serverRedirect)) {
return CompletableFuture.completedFuture(
new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, INVALID_SERVER_REDIRECT));
new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST,
Unpooled.copiedBuffer(INVALID_SERVER_REDIRECT, StandardCharsets.UTF_8)));
}
if (serverReference != null && serverReference.length() > MAX_SERVER_REFERENCE_LENGTH) {
return CompletableFuture.completedFuture(
new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, TOO_LONG_SERVER_REFERENCE));
new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST,
Unpooled.copiedBuffer(TOO_LONG_SERVER_REFERENCE, StandardCharsets.UTF_8)));
}

ServerRedirection serverRedirection = buildServerRedirection(serverRedirect, serverReference);
Expand Down
17 changes: 10 additions & 7 deletions bifromq-apiserver/src/main/java/org/apache/bifromq/apiserver/http/handler/PubHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@
import static org.apache.bifromq.apiserver.http.handler.utils.HeaderUtils.getHeader;

import com.google.protobuf.ByteString;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
Expand All @@ -46,6 +45,7 @@
import io.swagger.v3.oas.annotations.responses.ApiResponses;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
Expand All @@ -61,9 +61,9 @@

@Path("/pub")
final class PubHandler extends TenantAwareHandler {
private static final ByteBuf UNACCEPTED_TOPIC = Unpooled.wrappedBuffer("Unaccepted Topic".getBytes());
private static final ByteBuf INVALID_QOS = Unpooled.wrappedBuffer("Invalid QoS".getBytes());
private static final ByteBuf INVALID_EXPIRY_SECONDS = Unpooled.wrappedBuffer("Invalid expiry seconds".getBytes());
private static final String UNACCEPTED_TOPIC = "Unaccepted Topic";
private static final String INVALID_QOS = "Invalid QoS";
private static final String INVALID_EXPIRY_SECONDS = "Invalid expiry seconds";
private final IDistClient distClient;
private final ISettingProvider settingProvider;

Expand Down Expand Up @@ -114,15 +114,18 @@ public CompletableFuture<FullHttpResponse> handle(@Parameter(hidden = true) long
.orElse(Integer.MAX_VALUE);
if (!TopicUtil.checkTopicFilter(topic, tenantId, settingProvider)) {
return CompletableFuture.completedFuture(
new DefaultFullHttpResponse(req.protocolVersion(), FORBIDDEN, UNACCEPTED_TOPIC));
new DefaultFullHttpResponse(req.protocolVersion(), FORBIDDEN,
Unpooled.copiedBuffer(UNACCEPTED_TOPIC, StandardCharsets.UTF_8)));
}
if (qos < 0 || qos > 2) {
return CompletableFuture.completedFuture(
new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, INVALID_QOS));
new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST,
Unpooled.copiedBuffer(INVALID_QOS, StandardCharsets.UTF_8)));
}
if (expirySeconds <= 0) {
return CompletableFuture.completedFuture(
new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST, INVALID_EXPIRY_SECONDS));
new DefaultFullHttpResponse(req.protocolVersion(), BAD_REQUEST,
Unpooled.copiedBuffer(INVALID_EXPIRY_SECONDS, StandardCharsets.UTF_8)));
}
String clientType = getHeader(HEADER_CLIENT_TYPE, req, true);
Map<String, String> clientMeta = getClientMeta(req);
Expand Down
32 changes: 19 additions & 13 deletions ...mq-apiserver/src/test/java/org/apache/bifromq/apiserver/http/handler/KillHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,12 +35,12 @@
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;

import io.netty.handler.codec.http.DefaultFullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import org.apache.bifromq.sessiondict.client.ISessionDictClient;
import org.apache.bifromq.sessiondict.rpc.proto.KillReply;
Expand Down Expand Up @@ -254,12 +254,7 @@ public void invalidServerRedirect() {
long reqId = 123;
String tenantId = "bifromq_dev";

KillHandler handler = new KillHandler(settingProvider, sessionDictClient);
handler.handle(reqId, tenantId, req);
FullHttpResponse response = handler.handle(reqId, tenantId, req).join();
assertEquals(response.protocolVersion(), req.protocolVersion());
assertEquals(response.status(), HttpResponseStatus.BAD_REQUEST);
assertTrue(response.content().readableBytes() > 0);
assertRepeatedValidationResponse(reqId, tenantId, req, "Invalid server redirect value");
}

@Test
Expand All @@ -275,15 +270,26 @@ public void tooLongServerReference() {
long reqId = 123;
String tenantId = "bifromq_dev";

KillHandler handler = new KillHandler(settingProvider, sessionDictClient);
handler.handle(reqId, tenantId, req);
FullHttpResponse response = handler.handle(reqId, tenantId, req).join();
assertEquals(response.protocolVersion(), req.protocolVersion());
assertEquals(response.status(), HttpResponseStatus.BAD_REQUEST);
assertTrue(response.content().readableBytes() > 0);
assertRepeatedValidationResponse(reqId, tenantId, req, "Server reference exceeds 65535 bytes");
}

private DefaultFullHttpRequest buildRequest() {
return buildRequest(HttpMethod.DELETE);
}

private void assertRepeatedValidationResponse(long reqId, String tenantId, DefaultFullHttpRequest req,
String expectedContent) {
KillHandler handler = new KillHandler(settingProvider, sessionDictClient);
FullHttpResponse firstResponse = handler.handle(reqId, tenantId, req).join();
firstResponse.release();

FullHttpResponse secondResponse = handler.handle(reqId + 1, tenantId, req).join();
try {
assertEquals(secondResponse.protocolVersion(), req.protocolVersion());
assertEquals(secondResponse.status(), HttpResponseStatus.BAD_REQUEST);
assertEquals(secondResponse.content().toString(StandardCharsets.UTF_8), expectedContent);
} finally {
secondResponse.release();
}
}
}
49 changes: 48 additions & 1 deletion ...omq-apiserver/src/test/java/org/apache/bifromq/apiserver/http/handler/PubHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import org.mockito.Mock;
import org.testng.annotations.Test;
Expand Down Expand Up @@ -128,11 +129,42 @@ public void pubWithWrongQoS() {
}

@Test
public void pubWithWrongExpirySeconds() {
public void repeatedInvalidQoSResponseHasReadableContent() {
DefaultFullHttpRequest req = buildRequest();
req.headers().set(HEADER_TOPIC.header, "/greeting");
req.headers().set(HEADER_CLIENT_TYPE.header, "admin_team");
req.headers().set(HEADER_QOS.header, "3");

assertRepeatedValidationResponse(req, HttpResponseStatus.BAD_REQUEST, "Invalid QoS");
}

@Test
public void repeatedUnacceptedTopicResponseHasReadableContent() {
DefaultFullHttpRequest req = buildRequest();
req.headers().set(HEADER_TOPIC.header, "a".repeat(256));
req.headers().set(HEADER_CLIENT_TYPE.header, "admin_team");
req.headers().set(HEADER_QOS.header, "1");

assertRepeatedValidationResponse(req, HttpResponseStatus.FORBIDDEN, "Unaccepted Topic");
}

@Test
public void repeatedInvalidExpiryResponseHasReadableContent() {
DefaultFullHttpRequest req = buildRequest();
req.headers().set(HEADER_TOPIC.header, "/greeting");
req.headers().set(HEADER_CLIENT_TYPE.header, "admin_team");
req.headers().set(HEADER_QOS.header, "1");
req.headers().set(HEADER_EXPIRY_SECONDS.header, "0");

assertRepeatedValidationResponse(req, HttpResponseStatus.BAD_REQUEST, "Invalid expiry seconds");
}

@Test
public void pubWithWrongExpirySeconds() {
DefaultFullHttpRequest req = buildRequest();
req.headers().set(HEADER_TOPIC.header, "/greeting");
req.headers().set(HEADER_CLIENT_TYPE.header, "admin_team");
req.headers().set(HEADER_QOS.header, "1");
req.headers().set(HEADER_CLIENT_META_PREFIX + "age", "4");
req.headers().set(HEADER_EXPIRY_SECONDS.header, "0");
long reqId = 123;
Expand Down Expand Up @@ -195,4 +227,19 @@ public void tooLargeRequest() {
private DefaultFullHttpRequest buildRequest() {
return buildRequest(HttpMethod.POST);
}

private void assertRepeatedValidationResponse(DefaultFullHttpRequest req, HttpResponseStatus expectedStatus,
String expectedContent) {
PubHandler handler = new PubHandler(settingProvider, distClient);
FullHttpResponse firstResponse = handler.handle(123, "bifromq_dev", req).join();
firstResponse.release();

FullHttpResponse secondResponse = handler.handle(124, "bifromq_dev", req).join();
try {
assertEquals(secondResponse.status(), expectedStatus);
assertEquals(secondResponse.content().toString(StandardCharsets.UTF_8), expectedContent);
} finally {
secondResponse.release();
}
}
}
Loading