Merge branch 'master' into bugfix/adding-blocks-sample

Author: davidporter-id-au

Makefile

@@ -38,6 +38,7 @@ PROGS = helloworld \
 	sleep \
 	dataconverter \
 	autoscaling-monitoring \
+	batch \
 
 TEST_ARG ?= -race -v -timeout 5m
 BUILD := ./build
@@ -76,7 +77,7 @@ TEST_DIRS=./cmd/samples/cron \
 	./cmd/samples/recipes/dataconverter \
 	./cmd/samples/recovery \
 	./cmd/samples/pso \
-
+	./cmd/samples/batch \
 
 cancelactivity:
 	go build -o bin/cancelactivity cmd/samples/recipes/cancelactivity/*.go
@@ -192,6 +193,21 @@ dataconverter:
 autoscaling-monitoring:
 	go build -o bin/autoscaling-monitoring cmd/samples/advanced/autoscaling-monitoring/*.go
 
+batch:
+	go build -o bin/batch cmd/samples/batch/*.go
+
+test: bins
+	@rm -f test
+	@rm -f test.log
+	@echo $(TEST_DIRS)
+	@for dir in $(TEST_DIRS); do \
+		go test -coverprofile=$@ "$$dir" | tee -a test.log; \
+	done;
+
+clean:
+	rm -rf bin
+	rm -Rf $(BUILD)
+
 bins: helloworld \
 	blocks \
 	versioning \
@@ -226,15 +242,4 @@ bins: helloworld \
 	sleep \
 	dataconverter \
 	autoscaling-monitoring \
-
-test: bins
-	@rm -f test
-	@rm -f test.log
-	@echo $(TEST_DIRS)
-	@for dir in $(TEST_DIRS); do \
-		go test -coverprofile=$@ "$$dir" | tee -a test.log; \
-	done;
-
-clean:
-	rm -rf bin
-	rm -Rf $(BUILD)
+	batch \

cmd/samples/batch/README.md

@@ -0,0 +1,58 @@
+# Batch Processing Sample
+
+This sample demonstrates how to process large batches of activities with controlled concurrency using Cadence's `workflow.NewBatchFuture` functionality, while respecting the 1024 pending activities limit per workflow.
+
+## The problem it solves
+
+**The Problem**: When processing large datasets (thousands of records, files, or API calls), you face a dilemma:
+- **Sequential processing**: Too slow, poor user experience
+- **Unlimited concurrency**: Overwhelms databases, APIs, or downstream services
+- **Manual concurrency control**: Complex error handling and resource management
+- **Cadence limits**: Max 1024 pending activities per workflow
+
+**Real-world scenarios**:
+- Processing 10,000 user records for a migration
+- Sending emails to 50,000 subscribers
+- Generating reports for 1,000 customers
+- Processing files in a data pipeline
+
+### The Solution
+`workflow.NewBatchFuture` provides a robust solution:
+
+**Controlled Concurrency**: Process items in parallel while respecting system limits
+**Automatic Error Handling**: Failed activities don't crash the entire batch
+**Resource Efficiency**: Optimal throughput without overwhelming downstream services
+**Built-in Observability**: Monitoring, retries, and failure tracking
+**Workflow Integration**: Seamless integration with Cadence's workflow engine
+
+This eliminates the need to build custom concurrency control, error handling, and monitoring systems.
+
+## Sample behavior
+
+- Creates a configurable number of activities (default: 10)
+- Executes them with controlled concurrency (default: 3)
+- Simulates work with random delays (900-999ms per activity)
+- Handles cancellation gracefully
+
+## Technical considerations
+
+- **Cadence limit**: Maximum 1024 pending activities per workflow
+- **Resource management**: Controlled concurrency prevents system overload
+- **Error handling**: Failed activities don't crash the entire batch
+
+## How to run
+
+1. Build the sample:
+```bash
+make batch
+```
+
+2. Start Worker:
+```bash
+./bin/batch -m worker
+```
+
+3. Start Workflow:
+```bash
+./bin/batch -m trigger
+```

cmd/samples/batch/main.go

@@ -0,0 +1,71 @@
+package main
+
+import (
+	"flag"
+	"time"
+
+	"github.com/pborman/uuid"
+	"go.uber.org/cadence/client"
+	"go.uber.org/cadence/worker"
+
+	"github.com/uber-common/cadence-samples/cmd/samples/common"
+)
+
+// This needs to be done as part of a bootstrap step when the process starts.
+// The workers are supposed to be long running.
+func startWorkers(h *common.SampleHelper) {
+	// Configure worker options.
+	workerOptions := worker.Options{
+		MetricsScope: h.WorkerMetricScope,
+		Logger:       h.Logger,
+		FeatureFlags: client.FeatureFlags{
+			WorkflowExecutionAlreadyCompletedErrorEnabled: true,
+		},
+	}
+	h.StartWorkers(h.Config.DomainName, ApplicationName, workerOptions)
+}
+
+func startWorkflow(h *common.SampleHelper) {
+	workflowOptions := client.StartWorkflowOptions{
+		ID:                              "batch_" + uuid.New(),
+		TaskList:                        ApplicationName,
+		ExecutionStartToCloseTimeout:    time.Minute * 5,
+		DecisionTaskStartToCloseTimeout: time.Minute * 5,
+	}
+
+	// Default batch configuration
+	input := BatchWorkflowInput{
+		Concurrency: 3,
+		TotalSize:   10,
+	}
+
+	h.StartWorkflow(workflowOptions, batchWorkflowName, input)
+}
+
+func registerWorkflowAndActivity(
+	h *common.SampleHelper,
+) {
+	h.RegisterWorkflowWithAlias(BatchWorkflow, batchWorkflowName)
+	h.RegisterActivity(BatchActivity)
+}
+
+func main() {
+	var mode string
+	flag.StringVar(&mode, "m", "trigger", "Mode is worker or trigger.")
+	flag.Parse()
+
+	var h common.SampleHelper
+	h.SetupServiceConfig()
+
+	switch mode {
+	case "worker":
+		registerWorkflowAndActivity(&h)
+		startWorkers(&h)
+
+		// The workers are supposed to be long running process that should not exit.
+		// Use select{} to block indefinitely for samples, you can quit by CMD+C.
+		select {}
+	case "trigger":
+		startWorkflow(&h)
+	}
+}

cmd/samples/batch/workflow.go

@@ -1,4 +1,4 @@
-package batch
+package main
 
 import (
 	"context"
@@ -7,40 +7,50 @@ import (
 	"time"
 
 	"go.uber.org/cadence/workflow"
-	"go.uber.org/cadence/x"
 )
 
+// ApplicationName is the task list for this sample
+const ApplicationName = "batchGroup"
+
+const batchWorkflowName = "batchWorkflow"
+
 type BatchWorkflowInput struct {
 	Concurrency int
 	TotalSize   int
 }
 
 func BatchWorkflow(ctx workflow.Context, input BatchWorkflowInput) error {
+	// Create activity factories for each task (not yet executed)
 	factories := make([]func(workflow.Context) workflow.Future, input.TotalSize)
 	for taskID := 0; taskID < input.TotalSize; taskID++ {
-		taskID := taskID
+		taskID := taskID // Capture loop variable for closure
 		factories[taskID] = func(ctx workflow.Context) workflow.Future {
+			// Configure activity timeouts
 			aCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
-				ScheduleToStartTimeout: time.Second * 10,
-				StartToCloseTimeout:    time.Second * 10,
+				ScheduleToStartTimeout: time.Minute * 1,
+				StartToCloseTimeout:    time.Minute * 1,
 			})
 			return workflow.ExecuteActivity(aCtx, BatchActivity, taskID)
 		}
 	}
 
-	batch, err := x.NewBatchFuture(ctx, input.Concurrency, factories)
+	// Execute all activities with controlled concurrency
+	batch, err := workflow.NewBatchFuture(ctx, input.Concurrency, factories)
 	if err != nil {
 		return fmt.Errorf("failed to create batch future: %w", err)
 	}
 
+	// Wait for all activities to complete
 	return batch.Get(ctx, nil)
 }
 
 func BatchActivity(ctx context.Context, taskID int) error {
 	select {
 	case <-ctx.Done():
+		// Return error if workflow/activity is cancelled
 		return fmt.Errorf("batch activity %d failed: %w", taskID, ctx.Err())
 	case <-time.After(time.Duration(rand.Int63n(100))*time.Millisecond + 900*time.Millisecond):
+		// Wait for random duration (900-999ms) to simulate work, then return success
 		return nil
 	}
 }

cmd/samples/batch/workflow_test.go

@@ -1,4 +1,4 @@
-package batch
+package main
 
 import (
 	"testing"
@@ -8,17 +8,21 @@ import (
 )
 
 func Test_BatchWorkflow(t *testing.T) {
+	// Create test environment for workflow testing
 	testSuite := &testsuite.WorkflowTestSuite{}
 	env := testSuite.NewTestWorkflowEnvironment()
 
+	// Register the workflow and activity functions
 	env.RegisterWorkflow(BatchWorkflow)
 	env.RegisterActivity(BatchActivity)
 
+	// Execute workflow with 3 concurrent workers processing 10 tasks
 	env.ExecuteWorkflow(BatchWorkflow, BatchWorkflowInput{
-		Concurrency: 2,
+		Concurrency: 3,
 		TotalSize:   10,
 	})
 
+	// Assert workflow completed successfully without errors
 	assert.True(t, env.IsWorkflowCompleted())
 	assert.Nil(t, env.GetWorkflowError())
 }