You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2024/02/27 06:44:02 UTC
(pulsar) branch branch-3.2 updated: [bug][fn] Prevent `putstate` uses empty values (#22127)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.2 by this push:
new b5b402aee7b [bug][fn] Prevent `putstate` uses empty values (#22127)
b5b402aee7b is described below
commit b5b402aee7b3168786d4d61f71880180c74895fa
Author: Rui Fu <fr...@users.noreply.github.com>
AuthorDate: Tue Feb 27 14:42:52 2024 +0800
[bug][fn] Prevent `putstate` uses empty values (#22127)
---
.../functions/worker/rest/api/ComponentImpl.java | 18 +++++-----
.../integration/functions/PulsarStateTest.java | 42 ++++++++++++++++++++++
2 files changed, 52 insertions(+), 8 deletions(-)
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
index c7d227890cb..41b4c52c581 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
@@ -1235,15 +1235,17 @@ public abstract class ComponentImpl implements Component<PulsarWorkerService> {
try {
DefaultStateStore store = worker().getStateStoreProvider().getStateStore(tenant, namespace, functionName);
ByteBuffer data;
- if (state.getStringValue() != null) {
- data = ByteBuffer.wrap(state.getStringValue().getBytes(UTF_8));
- } else if (state.getByteValue() != null) {
- data = ByteBuffer.wrap(state.getByteValue());
- } else if (state.getNumberValue() != null) {
- data = ByteBuffer.allocate(Long.BYTES);
- data.putLong(state.getNumberValue());
+ if (state.getByteValue() == null || state.getByteValue().length == 0) {
+ if (state.getStringValue() != null) {
+ data = ByteBuffer.wrap(state.getStringValue().getBytes(UTF_8));
+ } else if (state.getNumberValue() != null) {
+ data = ByteBuffer.allocate(Long.BYTES);
+ data.putLong(state.getNumberValue());
+ } else {
+ throw new IllegalArgumentException("Invalid state value");
+ }
} else {
- throw new IllegalArgumentException("Invalid state value");
+ data = ByteBuffer.wrap(state.getByteValue());
}
store.put(key, data);
} catch (Throwable e) {
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarStateTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarStateTest.java
index f44cdffe639..856e4edfea0 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarStateTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarStateTest.java
@@ -28,6 +28,7 @@ import static org.testng.Assert.assertTrue;
import static org.testng.Assert.expectThrows;
import static org.testng.Assert.fail;
import com.google.common.base.Utf8;
+import com.google.gson.Gson;
import java.util.Base64;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
@@ -112,6 +113,20 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
String expectNumber = "\"numberValue\": 20";
putAndQueryState(functionName, "test-number", numberState, expectNumber);
+ byte[] valueBytes = Base64.getDecoder().decode(VALUE_BASE64);
+ String bytesString = Base64.getEncoder().encodeToString(valueBytes);
+ String byteState = "{\"key\":\"test-bytes\",\"byteValue\":\"" + bytesString + "\"}";
+ putAndQueryStateByte(functionName, "test-bytes", byteState, valueBytes);
+
+ String valueStr = "hello pulsar";
+ byte[] valueStrBytes = valueStr.getBytes(UTF_8);
+ String bytesStrString = Base64.getEncoder().encodeToString(valueStrBytes);
+ String byteStrState = "{\"key\":\"test-str-bytes\",\"byteValue\":\"" + bytesStrString + "\"}";
+ putAndQueryState(functionName, "test-str-bytes", byteStrState, valueStr);
+
+ String byteStrStateWithEmptyValues = "{\"key\":\"test-str-bytes\",\"byteValue\":\"" + bytesStrString + "\",\"stringValue\":\"\",\"numberValue\":0}";
+ putAndQueryState(functionName, "test-str-bytes", byteStrStateWithEmptyValues, valueStr);
+
// delete function
deleteFunction(functionName);
@@ -539,6 +554,33 @@ public class PulsarStateTest extends PulsarStandaloneTestSuite {
assertTrue(result.getStdout().contains(expect));
}
+ private void putAndQueryStateByte(String functionName, String key, String state, byte[] expect)
+ throws Exception {
+ container.execCmd(
+ PulsarCluster.ADMIN_SCRIPT,
+ "functions",
+ "putstate",
+ "--tenant", "public",
+ "--namespace", "default",
+ "--name", functionName,
+ "--state", state
+ );
+
+ ContainerExecResult result = container.execCmd(
+ PulsarCluster.ADMIN_SCRIPT,
+ "functions",
+ "querystate",
+ "--tenant", "public",
+ "--namespace", "default",
+ "--name", functionName,
+ "--key", key
+ );
+
+ FunctionState byteState = new Gson().fromJson(result.getStdout(), FunctionState.class);
+ assertNull(byteState.getStringValue());
+ assertEquals(byteState.getByteValue(), expect);
+ }
+
private void publishAndConsumeMessages(String inputTopic,
String outputTopic,
int numMessages) throws Exception {