From 773baad4228d38c0181f4a22c796a045473b596f Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Tue, 25 Aug 2026 08:32:17 -0700 Subject: [PATCH 1/3] [fix][fn] Allow retainKeyOrdering on Go functions ### Motivation #26414 taught the Go runtime to honour retainKeyOrdering: resolveSubscriptionType in pulsar-function-go/pf/instance.go selects a KeyShared subscription for it, the same way python_instance.py does. Nothing can reach that code on the cluster path. doGolangChecks still refuses retainKeyOrdering outright ("Retain Key Orderering not yet supported in Go function"), and validateNonJavaFunction is called by the worker REST API (FunctionsImpl), so cluster submission is exactly what the guard blocks. LocalRunner never calls it, which is why the KeyShared branch is reachable under localrun and nowhere else. Unlike the Python dead letter case, convert() carries retainKeyOrdering into FunctionDetails unconditionally, so this guard is the only thing in the way. ### Modifications Drop the guard from doGolangChecks. The combinations that would genuinely be contradictory are already rejected in doCommonChecks, for every runtime: retainKeyOrdering with EFFECTIVELY_ONCE ("retain Key ordering cannot be set"), and retainKeyOrdering together with retainOrdering ("Only one of retain ordering or retain key ordering can be set"). doGolangChecks refuses EFFECTIVELY_ONCE for Go before either is reached, so nothing needs repeating here. The Go client needs no extra configuration for KeyShared: ConsumerOptions leaves KeySharedPolicy nil, toProtoKeySharedMeta(nil) returns nil, and the broker then applies its default auto-split hash range. The maxMessageRetries guard stays: the Go runtime does not honour retryDetails yet, and a test pins that so this change cannot be widened by accident. ### Verifying this change Five tests in FunctionConfigUtilsTest: Go accepts retainKeyOrdering and converts to a KEY_SHARED subscription; accepts retainOrdering and converts to FAILOVER; rejects both ordering modes together; rejects retainKeyOrdering with EFFECTIVELY_ONCE; and still refuses message retries. Confirmed the tests are not vacuous: restoring the guard fails testGoFunctionAcceptsRetainKeyOrdering. --- .../functions/utils/FunctionConfigUtils.java | 8 ++- .../utils/FunctionConfigUtilsTest.java | 67 +++++++++++++++++++ 2 files changed, 72 insertions(+), 3 deletions(-) diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java index d88ac2820a966..9862acb7a849c 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java @@ -775,9 +775,11 @@ private static void doGolangChecks(FunctionConfig functionConfig) { throw new IllegalArgumentException("Message retries not yet supported in Go function"); } - if (functionConfig.getRetainKeyOrdering() != null && functionConfig.getRetainKeyOrdering()) { - throw new IllegalArgumentException("Retain Key Orderering not yet supported in Go function"); - } + // retainKeyOrdering is no longer refused: the Go runtime honours it, selecting a KeyShared + // subscription in resolveSubscriptionType (pulsar-function-go/pf/instance.go), the same way + // python_instance.py does. doCommonChecks still rejects the two combinations that would be + // contradictory -- retainKeyOrdering with EFFECTIVELY_ONCE, and retainKeyOrdering together + // with retainOrdering -- so nothing here needs to repeat them for Go. } private static void verifyNoTopicClash(Collection inputTopics, String outputTopic) diff --git a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java index ba5e40429cf2d..dd0eb8d98b709 100644 --- a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java +++ b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java @@ -19,6 +19,7 @@ package org.apache.pulsar.functions.utils; import static org.apache.pulsar.common.functions.FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE; +import static org.apache.pulsar.common.functions.FunctionConfig.Runtime.GO; import static org.apache.pulsar.common.functions.FunctionConfig.Runtime.PYTHON; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; @@ -29,6 +30,7 @@ import java.lang.reflect.Field; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.concurrent.atomic.AtomicReference; @@ -763,4 +765,69 @@ public void testConvertProducerSpecToProducerConfigAndBackToProducerSpec() { producerSpec.getCryptoSpec().getProducerEncryptionKeyNameAt(i)); } } + + private static FunctionConfig minimalGoFunctionConfig() { + FunctionConfig functionConfig = new FunctionConfig(); + functionConfig.setTenant("test-tenant"); + functionConfig.setNamespace("test-namespace"); + functionConfig.setName("test-function"); + functionConfig.setInputs(Collections.singletonList("persistent://public/default/input")); + functionConfig.setRuntime(GO); + functionConfig.setGo("/path/to/function"); + return functionConfig; + } + + @Test + public void testGoFunctionAcceptsRetainKeyOrdering() { + FunctionConfig functionConfig = minimalGoFunctionConfig(); + functionConfig.setRetainKeyOrdering(true); + + FunctionConfigUtils.validateNonJavaFunction(functionConfig); + + // The KeyShared subscription the Go runtime selects has to survive conversion, otherwise the + // instance never sees it. + assertEquals(FunctionConfigUtils.convert(functionConfig).getSource().getSubscriptionType(), + SubscriptionType.KEY_SHARED); + } + + @Test + public void testGoFunctionAcceptsRetainOrdering() { + FunctionConfig functionConfig = minimalGoFunctionConfig(); + functionConfig.setRetainOrdering(true); + + FunctionConfigUtils.validateNonJavaFunction(functionConfig); + + assertEquals(FunctionConfigUtils.convert(functionConfig).getSource().getSubscriptionType(), + SubscriptionType.FAILOVER); + } + + @Test(expectedExceptions = IllegalArgumentException.class, + expectedExceptionsMessageRegExp = "Only one of retain ordering or retain key ordering can be set") + public void testGoFunctionRejectsBothOrderingModes() { + FunctionConfig functionConfig = minimalGoFunctionConfig(); + functionConfig.setRetainOrdering(true); + functionConfig.setRetainKeyOrdering(true); + + FunctionConfigUtils.validateNonJavaFunction(functionConfig); + } + + @Test(expectedExceptions = IllegalArgumentException.class, + expectedExceptionsMessageRegExp = + "When effectively once processing guarantee is specified, retain Key ordering cannot be set") + public void testGoFunctionRejectsRetainKeyOrderingWithEffectivelyOnce() { + FunctionConfig functionConfig = minimalGoFunctionConfig(); + functionConfig.setRetainKeyOrdering(true); + functionConfig.setProcessingGuarantees(EFFECTIVELY_ONCE); + + FunctionConfigUtils.validateNonJavaFunction(functionConfig); + } + + @Test(expectedExceptions = IllegalArgumentException.class, + expectedExceptionsMessageRegExp = "Message retries not yet supported in Go function") + public void testGoFunctionStillRejectsMessageRetries() { + FunctionConfig functionConfig = minimalGoFunctionConfig(); + functionConfig.setMaxMessageRetries(3); + + FunctionConfigUtils.validateNonJavaFunction(functionConfig); + } } From 046e32b401486e82e26bf9ae56b928411b86a3e3 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Tue, 25 Aug 2026 08:32:18 -0700 Subject: [PATCH 2/3] [fix][fn] Correct the runtime markers on --retain-ordering and --retain-key-ordering The @Option descriptions in CmdFunctions carry a runtime marker that the docs sync parses into the "Support" column of the published pulsar-admin CLI reference, and that `functions create --help` prints verbatim. Both flags were marked #Java. That has been wrong for Python since long before #26414: python_instance.py selects Failover for retainOrdering and KeyShared for retainKeyOrdering. #26414 added the same to the Go runtime, and the preceding commit makes retainKeyOrdering reachable for a Go function submitted to a cluster, so both now read #Java, Python, Go. --- .../main/java/org/apache/pulsar/admin/cli/CmdFunctions.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java index ce3d8d323685f..889d90e76e95d 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java @@ -297,10 +297,11 @@ abstract class FunctionDetailsCommand extends BaseCommand { @Option(names = "--retainOrdering", description = "Function consumes and processes messages in order", hidden = true) protected Boolean deprecatedRetainOrdering; - @Option(names = "--retain-ordering", description = "Function consumes and processes messages in order #Java") + @Option(names = "--retain-ordering", + description = "Function consumes and processes messages in order #Java, Python, Go") protected Boolean retainOrdering; @Option(names = "--retain-key-ordering", - description = "Function consumes and processes messages in key order #Java") + description = "Function consumes and processes messages in key order #Java, Python, Go") protected Boolean retainKeyOrdering; @Option(names = "--batch-builder", description = "BatcherBuilder provides two types of " + "batch construction methods, DEFAULT and KEY_BASED. The default value is: DEFAULT") From 3beffb49e1fffa2d07dfc77a5314fdfb984da7d9 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 26 Aug 2026 10:21:49 -0700 Subject: [PATCH 3/3] [fix][fn] Drop the explanatory block from doGolangChecks The active constraints are already stated in doCommonChecks, which rejects retainKeyOrdering with EFFECTIVELY_ONCE and retainKeyOrdering together with retainOrdering, and both are covered by tests. Keeping doGolangChecks limited to active validation avoids a second description of the rules, and drops references to specific Go and Python implementation files that would go stale. --- .../apache/pulsar/functions/utils/FunctionConfigUtils.java | 6 ------ 1 file changed, 6 deletions(-) diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java index 9862acb7a849c..81a3f50508cdd 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java @@ -774,12 +774,6 @@ private static void doGolangChecks(FunctionConfig functionConfig) { if (functionConfig.getMaxMessageRetries() != null && functionConfig.getMaxMessageRetries() >= 0) { throw new IllegalArgumentException("Message retries not yet supported in Go function"); } - - // retainKeyOrdering is no longer refused: the Go runtime honours it, selecting a KeyShared - // subscription in resolveSubscriptionType (pulsar-function-go/pf/instance.go), the same way - // python_instance.py does. doCommonChecks still rejects the two combinations that would be - // contradictory -- retainKeyOrdering with EFFECTIVELY_ONCE, and retainKeyOrdering together - // with retainOrdering -- so nothing here needs to repeat them for Go. } private static void verifyNoTopicClash(Collection inputTopics, String outputTopic)