Fix: set subscription-type based on message ordering (apache#2259)

* Fix: set subscription-type based on message ordering * set failover sub on EFFECTIVELY_ONCE processing guarantee
nahguam · Aug 2, 2018 · c85dc46 · c85dc46
1 parent d2f6dd9
commit c85dc46
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 22 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
@@ -86,6 +86,7 @@
 import org.apache.pulsar.functions.utils.Reflections;
 import org.apache.pulsar.functions.utils.Utils;
 import org.apache.pulsar.functions.utils.WindowConfig;
+import org.apache.pulsar.functions.utils.FunctionConfig.ProcessingGuarantees;
 import org.apache.pulsar.functions.utils.validation.ConfigValidation;
 import org.apache.pulsar.functions.utils.validation.ValidatorImpls.ImplementsClassesValidator;
 import org.apache.pulsar.functions.windowing.WindowFunctionExecutor;
@@ -240,6 +241,8 @@ abstract class FunctionDetailsCommand extends BaseCommand {
         protected FunctionConfig.ProcessingGuarantees processingGuarantees;
         @Parameter(names = "--userConfig", description = "User-defined config key/values")
         protected String userConfigString;
+        @Parameter(names = "--retainOrdering", description = "Function consumes and processes messages in order")
+        protected boolean retainOrdering;
         @Parameter(names = "--parallelism", description = "The function's parallelism factor (i.e. the number of function instances to run)")
         protected Integer parallelism;
         @Parameter(names = "--cpu", description = "The cpu in cores that need to be allocated per function instance(applicable only to docker runtime)")
@@ -315,6 +318,9 @@ void processArguments() throws Exception {
             if (null != processingGuarantees) {
                 functionConfig.setProcessingGuarantees(processingGuarantees);
             }
+
+            functionConfig.setRetainOrdering(retainOrdering);
+
             if (null != userConfigString) {
                 Type type = new TypeToken<Map<String, String>>(){}.getType();
                 Map<String, Object> userConfigMap = new Gson().fromJson(userConfigString, type);
@@ -567,24 +573,13 @@ protected FunctionDetails convert(FunctionConfig functionConfig)
                 sourceSpecBuilder.setTopicsPattern(functionConfig.getTopicsPattern());
             }
 
-            // Set subscription type based on processing semantics
-            if (functionConfig.getProcessingGuarantees() != null) {
-                switch (functionConfig.getProcessingGuarantees()) {
-                    case ATMOST_ONCE:
-                        sourceSpecBuilder.setSubscriptionType(SubscriptionType.SHARED);
-                        break;
-                    case ATLEAST_ONCE:
-                        sourceSpecBuilder.setSubscriptionType(SubscriptionType.SHARED);
-                        break;
-                    case EFFECTIVELY_ONCE:
-                        sourceSpecBuilder.setSubscriptionType(SubscriptionType.FAILOVER);
-                        break;
-                    default:
-                        throw new RuntimeException("Unknown processing guarantee: "
-                                + functionConfig.getProcessingGuarantees().name());
-                }
-            }
-
+            // Set subscription type based on ordering and EFFECTIVELY_ONCE semantics
+            SubscriptionType subType = (functionConfig.isRetainOrdering()
+                    || ProcessingGuarantees.EFFECTIVELY_ONCE.equals(functionConfig.getProcessingGuarantees()))
+                            ? SubscriptionType.FAILOVER
+                            : SubscriptionType.SHARED;
+            sourceSpecBuilder.setSubscriptionType(subType);
+
             if (typeArgs != null) {
                 sourceSpecBuilder.setTypeClassName(typeArgs[0].getName());
             }

diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
@@ -57,6 +57,7 @@
 import org.apache.pulsar.functions.proto.Function.SourceSpec;
 import org.apache.pulsar.functions.proto.Function.SubscriptionType;
 import org.apache.pulsar.functions.utils.FunctionConfig;
+import org.apache.pulsar.functions.utils.FunctionConfig.ProcessingGuarantees;
 import org.apache.pulsar.functions.utils.SinkConfig;
 import org.apache.pulsar.functions.utils.Utils;
 import org.apache.pulsar.functions.utils.io.ConnectorUtils;
@@ -468,7 +469,6 @@ protected FunctionDetails createSinkConfig(SinkConfig sinkConfig) throws IOExcep
             // set source spec
             // source spec classname should be empty so that the default pulsar source will be used
             SourceSpec.Builder sourceSpecBuilder = SourceSpec.newBuilder();
-            sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.SHARED);
             if (sinkConfig.getTopicToSerdeClassName() != null) {
                 sourceSpecBuilder.putAllTopicsToSerDeClassName(sinkConfig.getTopicToSerdeClassName());
             }
@@ -483,9 +483,12 @@ protected FunctionDetails createSinkConfig(SinkConfig sinkConfig) throws IOExcep
                 sourceSpecBuilder.setSubscriptionName(sinkConfig.getSourceSubscriptionName());
             }
 
-            sourceSpecBuilder.setSubscriptionType(
-                    sinkConfig.isRetainOrdering() ? SubscriptionType.FAILOVER : SubscriptionType.SHARED);
-
+            SubscriptionType subType = (sinkConfig.isRetainOrdering()
+                    || ProcessingGuarantees.EFFECTIVELY_ONCE.equals(sinkConfig.getProcessingGuarantees()))
+                            ? SubscriptionType.FAILOVER
+                            : SubscriptionType.SHARED;
+            sourceSpecBuilder.setSubscriptionType(subType);
+
             functionDetailsBuilder.setAutoAck(true);
             functionDetailsBuilder.setSource(sourceSpecBuilder);
 

diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfig.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfig.java
@@ -85,6 +85,7 @@ public enum Runtime {
     @isValidTopicName
     private String logTopic;
     private ProcessingGuarantees processingGuarantees;
+    private boolean retainOrdering;
     private Map<String, Object> userConfig;
     private Runtime runtime;
     private boolean autoAck;