Skip to content
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

[fix][fn] Prevent create state table from API calls for non-exists instances #22107

Merged
merged 1 commit into from
Feb 25, 2024
Merged

[fix][fn] Prevent create state table from API calls for non-exists instances #22107

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
12 changes: 12 additions & 0 deletions ...tions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1150,6 +1150,12 @@ public FunctionState getFunctionState(final String tenant,
throw new RestException(Status.BAD_REQUEST, e.getMessage());
}

FunctionMetaDataManager functionMetaDataManager = worker().getFunctionMetaDataManager();
if (!functionMetaDataManager.containsFunction(tenant, namespace, functionName)) {
log.warn("getFunctionState does not exist @ /{}/{}/{}", tenant, namespace, functionName);
throw new RestException(Status.NOT_FOUND, String.format("'%s' is not found", functionName));
}

try {
DefaultStateStore store = worker().getStateStoreProvider().getStateStore(tenant, namespace, functionName);
StateValue value = store.getStateValue(key);
Expand Down Expand Up @@ -1219,6 +1225,12 @@ public void putFunctionState(final String tenant,
throw new RestException(Status.BAD_REQUEST, e.getMessage());
}

FunctionMetaDataManager functionMetaDataManager = worker().getFunctionMetaDataManager();
if (!functionMetaDataManager.containsFunction(tenant, namespace, functionName)) {
log.warn("putFunctionState does not exist @ /{}/{}/{}", tenant, namespace, functionName);
throw new RestException(Status.NOT_FOUND, String.format("'%s' is not found", functionName));
}

try {
DefaultStateStore store = worker().getStateStoreProvider().getStateStore(tenant, namespace, functionName);
ByteBuffer data;
Expand Down
30 changes: 30 additions & 0 deletions ...egration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarStateTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,14 @@ public void testSourceState() throws Exception {
assertEquals(e.getStatusCode(), 404);
}

// query a non-exist instance should get a 404 error
{
PulsarAdminException e = expectThrows(PulsarAdminException.class, () -> {
admin.functions().getFunctionState("public", "default", "non-exist", "non-exist");
});
assertEquals(e.getStatusCode(), 404);
}

Awaitility.await().ignoreExceptions().untilAsserted(() -> {
FunctionState functionState = admin.functions().getFunctionState("public", "default", sourceName, "now");
assertTrue(functionState.getStringValue().matches("val1-.*"));
Expand Down Expand Up @@ -204,6 +212,14 @@ public void testSinkState() throws Exception {
assertEquals(e.getStatusCode(), 404);
}

// query a non-exist instance should get a 404 error
{
PulsarAdminException e = expectThrows(PulsarAdminException.class, () -> {
admin.functions().getFunctionState("public", "default", "non-exist", "non-exist");
});
assertEquals(e.getStatusCode(), 404);
}

for (int i = 0; i < numMessages; i++) {
producer.send("foo");
}
Expand All @@ -226,6 +242,20 @@ public void testSinkState() throws Exception {
getSinkInfoNotFound(sinkName);
}

@Test(groups = {"python_state", "state", "function", "python_function"})
public void testNonExistFunction() throws Exception {
String functionName = "non-exist-function-" + randomName(8);
try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(container.getHttpServiceUrl()).build()) {
// query a non-exist instance should get a 404 error
{
PulsarAdminException e = expectThrows(PulsarAdminException.class, () -> {
admin.functions().getFunctionState("public", "default", functionName, "non-exist");
});
assertEquals(e.getStatusCode(), 404);
}
}
}

@Test(groups = {"java_state", "state", "function", "java_function"})
public void testBytes2StringNotUTF8() {
byte[] valueBytes = Base64.getDecoder().decode(VALUE_BASE64);
Expand Down
Loading