Add foreach flattened view support via maestro-extensions module

[rdeepak2002]

Pull Request type

• Bugfix
• Feature
• Refactoring (no functional changes, no api changes)
• Build related changes (Please run ./gradlew build --write-locks to refresh dependencies)
• Other (please describe):

NOTE: Please remember to run ./gradlew spotlessApply to fix any format violations.

Changes in this PR

Ports the "flattened foreach" feature from internal Maestro (maestro-signal-service) to OSS as a new maestro-extensions module. This is an auto-configurable library that registers into maestro-server when extensions.enabled=true (set in the aws profile). It consumes maestro events via SQS (subscribed to the SNS topic) and makes HTTP calls to maestro-server for data — matching the internal architecture.

The denormalization strategy stores foreach step iterations in a dedicated maestro_step_foreach_flattened table for efficient cursor-based pagination, status filtering, and summary statistics — critical for workflows with large iteration counts.

Architecture:

• maestro-extensions is an auto-configurable library registered into maestro-server via Spring Boot auto-configuration (all endpoints on port 8080)
• Uses @ConditionalOnProperty(value = "extensions.enabled", havingValue = "true") to activate only when configured
• Uses @ComponentScan to register REST controllers from the extensions package
• Injects shared beans from the host application: DataSource, ObjectMapper, MaestroMetrics, DatabaseConfiguration, HttpClient
• Consumes StepInstanceStatusChangeEvent from an SQS queue (maestro-event) subscribed to the maestro SNS topic
• Makes HTTP GET calls to maestro-server REST API via HttpMaestroClient for workflow/step data
• Flyway migrations auto-discovered at classpath:db/migration/postgres/ when jar is on classpath
• Manual SQS acknowledgment with DLQ routing (redrive after 5 failures), visibility timeout on failure, and receive count monitoring — matching internal's SqsProcessorFinalizer pattern
• SNS→SQS subscription uses RawMessageDelivery=true so messages arrive as raw JSON (matching internal)

---

File inventory

Files matching internal (logic identical, only package/annotation/import differences)

These files are direct ports from maestro-signal-service. The core logic is identical; differences are limited to package names, license headers, DI annotations (@Inject/@Singleton → Spring @Bean), metrics class names (MetricRepo → MaestroMetrics), and @SuppressFBWarnings annotations.

OSS file	Internal counterpart	Notes
handlers/ForeachFlatteningHandler.java	handlers/foreach/ForeachFlatteningHandler.java	+ @SuppressFBWarnings
processors/StepEventPreprocessor.java	messageProcessors/StepEventPreprocessor.java	MetricRepo → MaestroMetrics; metric constant inlined
dao/models/ForeachFlattenedInstance.java	dao/models/ForeachFlattenedInstance.java	Package only
dao/models/ForeachFlattenedModel.java	dao/models/ForeachFlattenedModel.java	Package only
models/StepEventHandlerInput.java	models/StepEventHandlerInput.java	+ @SuppressFBWarnings
models/StepIteration.java	client/models/flattening/StepIteration.java	Package only
models/StepIterationsSummary.java	client/models/flattening/StepIterationsSummary.java	Package only
utils/ForeachFlatteningHelper.java	utils/ForeachFlatteningHelper.java	Package only
utils/StepInstanceStatusEncoder.java	utils/StepInstanceStatusEncoder.java	EnumMap return/field types → Map interface (PMD LooseCoupling rule); internal logic identical
controllers/ForeachFlattenController.java	controllers/flattening/ForeachFlattenController.java	@SuppressWarnings → @SuppressFBWarnings; javax.annotation.Nullable → com.netflix.maestro.annotations.Nullable

Files with logical differences from internal

OSS file	Internal counterpart	Differences
processors/MaestroEventProcessor.java	messageProcessors/MaestroEventProcessor.java	Removes TargetedAtGroupHandler, TrinoDirectHandler, and WORKFLOW_INSTANCE_STATUS_CHANGE_EVENT handling (not in scope for OSS). Changes MessageProcessor<T> interface with Supplier<MaestroEvent> to direct MaestroEvent parameter. Core foreach-handling logic is identical.
listeners/SqsMaestroEventListener.java	listeners/SqsMaestroEventListener.java	Internal delegates to shared SqsProcessorFinalizer utility class; OSS inlines the same manual ack / visibility / DLQ logic directly in the listener (same behavior, no shared library abstraction). Uses ObjectMapper.readValue() directly instead of MessageProcessor<T> with Supplier<T>.
dao/MaestroForeachFlattenedDao.java	dao/flattening/MaestroForeachFlattenedDao.java	Internal supports dual CockroachDB + Postgres with a runtime feature flag (SignalFastProperties.isUsePostgresForSignalService()), two separate INSERT queries, and @iterationIdx CRDB index hints. OSS is Postgres-only — single INSERT query, no CRDB conditionals.
utils/PaginationHelper.java	shared/utils/PaginationHelper.java	OSS keeps only validateParamAndDeriveDirection() + buildEmptyPaginationResult() (used by the controller). Internal's full pagination range calculation and result building are for other signal-service features not ported here.

OSS-only files (no internal counterpart)

These files are needed because OSS uses Spring Boot (vs internal's Micronaut) and doesn't have access to internal shared libraries (maestro-client, maestro-shared).

OSS file	Purpose
config/MaestroExtensionsConfiguration.java	Spring @Configuration with @ConditionalOnProperty + @ComponentScan for auto-config. Injects host beans via @Qualifier(Constants.MAESTRO_QUALIFIER) and parameter naming.
properties/MaestroExtensionsProperties.java	@ConfigurationProperties for SQS queue URL, maestro base URL
provider/MaestroClient.java	Interface abstracting data fetching — matches internal's MaestroClient method names (getWorkflowInstance, getWorkflowDefinition, getWorkflowStepInstance). Internal's MaestroClient is an 800+ line concrete class with SSLSocketFactory, streaming iterators, POST methods; OSS extracts only the 3 methods needed for foreach flattening.
provider/HttpMaestroClient.java	JDK HttpClient implementation calling maestro-server REST API (internal's MaestroClient uses Metatron mTLS)

Resource files

File	Description
src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports	Spring Boot auto-configuration registration
src/main/resources/db/migration/postgres/V202311161000__add_foreach_flattened.sql	Flyway migration for maestro_step_foreach_flattened table (identical to internal, trailing newline only)
src/test/resources/flattening/flattening_records.json	Test fixture for DAO tests (identical to internal)

Test files (11 files)

File	Covers	Notes
ExtensionsBaseTest.java	Shared ObjectMapper setup	OSS-only (replaces internal's SignalBaseTest/MaestroBaseTest)
ExtensionsDaoBaseTest.java	Testcontainers Postgres + Flyway + DataSource	OSS-only (replaces internal's SignalDaoBaseTest)
MaestroForeachFlattenedDaoTest.java	DAO insert/query/summary with real Postgres (11 tests)	Same 11 tests as internal; removes CRDB code paths (SignalFastProperties mock, dual INSERT)
ForeachFlatteningHandlerTest.java	Handler logic (2 tests)	Matches internal (package/import only)
ForeachFlatteningHelperTest.java	Model building logic (5 tests)	Matches internal (package/import only)
StepInstanceStatusEncoderTest.java	Status encoding/decoding (2 tests)	Matches internal (package only)
ForeachFlattenControllerTest.java	REST endpoints with mocked DAO (9 tests)	Matches internal (package/import only)
StepEventPreprocessorTest.java	Preprocessor (2 tests)	Same 2 scenarios as internal; rewritten with StepInstanceStatusChangeEvent.builder() + AssertJ (internal uses mock fields + JUnit Assert + manual Spectator registry metric verification)
MaestroEventProcessorTest.java	Event dispatch + retryable error + metrics (4 tests)	4 tests vs internal's 7 — removed 3 targeted-at-group tests; kept foreach-relevant tests; rewritten with builder pattern
SqsMaestroEventListenerTest.java	Manual ack, deserialization error deletion, visibility extension (5 tests)	Covers same scenarios as internal's SqsProcessorFinalizerTest but tests the inlined listener logic
HttpMaestroClientTest.java	HTTP client with mocked responses (6 tests)	OSS-only (no internal counterpart)

Changes to existing modules

File	Change
settings.gradle	Added maestro-extensions to multi-project build
maestro-database/.../AbstractDatabaseDao.java	Added markTransactionSerializable()
maestro-aws/localstack/sqs_bootstrap.sh	Added maestro-event queue with DLQ + redrive policy
maestro-aws/localstack/sns_bootstrap.sh	Added SNS→SQS subscription with RawMessageDelivery=true
maestro-server/build.gradle	Added implementation project(':maestro-extensions') dependency
maestro-server/.../application-aws.yml	Added extensions config block (extensions.enabled, queue URL, base URL)
maestro-server/.../docker-java.properties	Testcontainers Docker API version fix

---

E2E Testing

These commands have been verified end-to-end on macOS. Copy and paste each command in order.

Prerequisites

• Docker Desktop running
• Java 21+

Step 1: Start infrastructure (LocalStack + Redis)

docker compose -f maestro-aws/docker-compose.yml up -d

Wait until docker ps shows both containers as healthy/running.

Step 2: Start maestro-server with extensions (single terminal)

./gradlew :maestro-server:bootRun --args='--spring.profiles.active=aws'

Wait for: Started MaestroApp in X seconds. Extensions auto-configure and you should see log lines like:

Creating HttpMaestroClient within Spring boot...
Creating MaestroForeachFlattenedDao within Spring boot...
Creating SqsMaestroEventListener within Spring boot...

Step 3: Create the foreach workflow

curl -s -X POST 'http://localhost:8080/api/v3/workflows' \
  -H 'Content-Type: application/json' \
  -H 'user: tester' \
  -d @maestro-server/src/test/resources/samples/sample-foreach-wf.json

Expected: JSON response with workflow_definition containing sample-foreach-wf.

Step 4: Trigger the workflow

curl -s -X POST 'http://localhost:8080/api/v3/workflows/sample-foreach-wf/versions/latest/actions/start' \
  -H 'Content-Type: application/json' \
  -H 'user: tester' \
  -d '{"initiator": {"type": "manual"}}'

Expected: JSON with "status":"CREATED" and workflow_instance_id: 1.

Step 5: Wait for execution + event processing

sleep 30

The foreach creates 9 iterations (dates 20200101-20200109), each with 6 inner steps (job.1-job.6). Events flow: maestro-server → SNS → SQS → extensions listener (same process).

Step 6: Check workflow completed

curl -s 'http://localhost:8080/api/v3/workflows/sample-foreach-wf/instances/1/runs/1' | python3 -m json.tool | grep status

Expected: "status": "SUCCEEDED"

Step 7: Query flattened foreach summary (same port 8080)

The flattened view is keyed by inner step ID (e.g., job.1), not the foreach step name.

curl -s 'http://localhost:8080/api/v3/views/flatten/workflows/sample-foreach-wf/instances/1/runs/1/steps/job.1/summary?statusCount=true' | python3 -m json.tool

Expected:

{
    "representative_iteration": { "step_id": "job.1", "loop_params": {"i": "20200101"}, "step_runtime_state": {"status": "SUCCEEDED"} },
    "count_by_status": { "SUCCEEDED": 9 },
    "loop_param_values": { "i": ["20200101", "20200102", "20200103", "20200104", "20200105"] }
}

Step 8: Query paginated iterations

curl -s 'http://localhost:8080/api/v3/views/flatten/workflows/sample-foreach-wf/instances/1/runs/1/steps/job.1?first=3' | python3 -m json.tool

Expected: 3 iterations with has_next_page: true, each showing loop param i and SUCCEEDED status.

Step 9: Get a single iteration by ID

curl -s 'http://localhost:8080/api/v3/views/flatten/workflows/sample-foreach-wf/instances/1/runs/1/steps/job.1/iterations/1' | python3 -m json.tool

Expected: Single iteration with iteration_rank: "11", loop_params: {"i": "20200101"}, status SUCCEEDED.

Cleanup

docker compose -f maestro-aws/docker-compose.yml down