[CDAP-21096] Remove in-memory launching queue from RunRecordMonitorService #15800

vsethi09 · 2025-01-13T12:37:41Z

Context

Multiple instances of RunRecordMonitoringService cannot run as distributed services as the in-memory cache of the launching queue will result in inconsistencies.

Removed the in-memory launching queue from the RunRecordMonitoringService and used AppMetadataStore APIs.

For more context, see: #15773 (comment).

Note: RunRecordMonitoringService is renamed as FlowControlMonitoringService.

Testing

Unit Tests
CDAP sandbox
Docker image

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/services/FlowControlMonitorService.java

tivv · 2025-01-15T21:27:30Z

...c/src/main/java/io/cdap/cdap/internal/app/services/ProgramNotificationSubscriberService.java

                    if (runRecordDetail.getStatus() == ProgramRunStatus.PENDING) {
-                      runRecordMonitorService.addRequest(runRecordDetail.getProgramRunId());
+                      flowControlMonitorService.addRequest(runRecordDetail.getProgramRunId(),


I don't think this is needed if the run is already in the database.

This is required for emitting flow control metrics.

Replaced addRequest() with emitFlowControlMetrics() call.

Well, originally we did it to recreate the in-memory store from database store. Now we read a database store. I don't see a reason to imeddiately write back. Do I miss something?

When active runs are restored (in restoreActiveRuns), the launching count metrics needs to be updated for runs in Pending state, otherwise metrics will show incorrect value.

This is only for consistency of launching count metrics.

Even if we don't update the launching count metrics, and a run request is made, it will update the DB and the launching count metric.

Metrics we already emit in the end of startup once, no need to do it for every run

Removed.

Yes, metrics are emitted [here](

cdap/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/services/ProgramNotificationSubscriberService.java

Line 162 in 66da732

emitFlowControlMetrics();

after all launches are restored.

So, coming back, why do you need to call flowControlService.addRequest? Run is already in the database, no need to add it again

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/store/AppMetadataStore.java

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/services/FlowControlService.java

+      int runningCount = getRunningCount(store);
+      return new Counter(launchingCount, runningCount);
+    });
+    LOG.info("Added request with runId {}.", programRunId);


tivv · 2025-01-21T22:26:37Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/services/FlowControlService.java

+   *
+   * @return Counter with total number of launching and running program runs.
+   */
+  public Counter getCount() {


nit: consider using plural (getCounts) or object name ('getCounter)

Renamed to getCounter().

Similarly renamed addRequestAndGetCount() to addRequestAndGetCounter()

tivv · 2025-01-21T22:27:50Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/services/FlowControlService.java

+     * * total number of run records with {@link ProgramRunStatus#PENDING} status + total number of
+     * run records with {@link ProgramRunStatus#STARTING} status.
+     */
+    private final Integer launchingCount;


Since both are always filled, can be primitives now

Changed to int.

tivv · 2025-01-21T22:30:42Z

...c/src/main/java/io/cdap/cdap/internal/app/services/ProgramNotificationSubscriberService.java

                    if (runRecordDetail.getStatus() == ProgramRunStatus.PENDING) {
-                      runRecordMonitorService.addRequest(runRecordDetail.getProgramRunId());
+                      flowControlMonitorService.addRequest(runRecordDetail.getProgramRunId(),


So, coming back, why do you need to call flowControlService.addRequest? Run is already in the database, no need to add it again

tivv · 2025-01-21T22:31:26Z

...c/src/main/java/io/cdap/cdap/internal/app/services/ProgramNotificationSubscriberService.java

+                    ProgramDescriptor programDescriptor = this.store.loadProgram(
+                        runRecordDetail.getProgramRunId().getParent());
+                    if (runRecordDetail.getStatus() == ProgramRunStatus.STARTING) {
+                      flowControlService.addRequest(runRecordDetail.getProgramRunId(),


I don't think you need this. Run is already in the database.

tivv · 2025-01-21T22:37:14Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/store/AppMetadataStore.java

@@ -1167,6 +1254,7 @@ public RunRecordDetail recordProgramOrphaned(ProgramRunId programRunId, long end
        .setCluster(cluster)
        .setSourceId(sourceId)
        .build();
+    addFlowControlStatusKey(key, meta);
    writeToStructuredTableWithPrimaryKeys(
        key, meta, getRunRecordsTable(), StoreDefinition.AppMetadataStore.RUN_RECORD_DATA);


nit: create a helper method for this two calls. (or 3 with LOG.trace). This would ensure adding status keys won't be missed and reduce duplicated code

Added 2 liner helper function: writeRunRecordWithPrimaryKeys().

Removed: addFlowControlStatusKey() and merged it into writeRunRecordWithPrimaryKeys().

Not adding the LOG.trace as at some places the program run status is logged while at some places cluster status if logged.

tivv · 2025-01-21T22:39:41Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/store/AppMetadataStore.java

@@ -150,6 +151,16 @@ public class AppMetadataStore {
      .put(ProgramRunStatus.REJECTED, TYPE_RUN_RECORD_COMPLETED)
      .build();

+  private static final String TYPE_FLOW_CONTROL_LAUNCHING = "flowControlLaunching";


Consider not using prefix in values. This would save database space and generally it's clear that it's flow control from field name. Also do you want to differentiate NONE from NULL? Most run records would be in NONE state, so using null would save quite some space. Also assume we don't do migrations, so counters would be off for any programs running while upgrading and that's fine (there should not be any). But having NULL would save you from updateing all old runs, all those would be consistent

sonarqubecloud · 2025-01-22T11:50:49Z

Quality Gate failed

Failed conditions
B Security Rating on New Code (required ≥ A)
C Reliability Rating on New Code (required ≥ A)

See analysis details on SonarQube Cloud

Catch issues before they fail your Quality Gate with our IDE extension SonarQube for IDE

vsethi09 added the build Triggers github actions build label Jan 13, 2025

vsethi09 force-pushed the feature/CDAP-21096_fix_RunRecordMonitorService_queue_cache branch 5 times, most recently from dc068c3 to 0ea30c6 Compare January 15, 2025 20:37

vsethi09 changed the title ~~[WIP] Remove in-memory launching queue in RunRecordMonitorService~~ [CDAP-21096] Remove in-memory launching queue from RunRecordMonitorService Jan 15, 2025

vsethi09 marked this pull request as ready for review January 15, 2025 20:42

vsethi09 force-pushed the feature/CDAP-21096_fix_RunRecordMonitorService_queue_cache branch from 0ea30c6 to abcd4c5 Compare January 15, 2025 20:49

tivv reviewed Jan 15, 2025

View reviewed changes

vsethi09 force-pushed the feature/CDAP-21096_fix_RunRecordMonitorService_queue_cache branch 2 times, most recently from 343101b to db8a3ab Compare January 17, 2025 15:24

vsethi09 requested a review from tivv January 17, 2025 15:26

vsethi09 force-pushed the feature/CDAP-21096_fix_RunRecordMonitorService_queue_cache branch from db8a3ab to 66da732 Compare January 17, 2025 16:07

github-advanced-security bot found potential problems Jan 17, 2025

View reviewed changes

vsethi09 force-pushed the feature/CDAP-21096_fix_RunRecordMonitorService_queue_cache branch 3 times, most recently from ea09155 to d9e4c13 Compare January 20, 2025 16:35

tivv reviewed Jan 21, 2025

View reviewed changes

Remove in-memory launching queue in RunRecordMonitorService

31c5b1c

vsethi09 force-pushed the feature/CDAP-21096_fix_RunRecordMonitorService_queue_cache branch from d9e4c13 to 31c5b1c Compare January 22, 2025 10:32

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[CDAP-21096] Remove in-memory launching queue from RunRecordMonitorService #15800

[CDAP-21096] Remove in-memory launching queue from RunRecordMonitorService #15800

vsethi09 commented Jan 13, 2025 •

edited

Loading

tivv Jan 15, 2025

vsethi09 Jan 16, 2025 •

edited

Loading

tivv Jan 16, 2025

vsethi09 Jan 17, 2025

tivv Jan 17, 2025

vsethi09 Jan 20, 2025

tivv Jan 21, 2025

tivv Jan 21, 2025

vsethi09 Jan 22, 2025

tivv Jan 21, 2025

vsethi09 Jan 22, 2025

tivv Jan 21, 2025

tivv Jan 21, 2025

vsethi09 Jan 22, 2025

tivv Jan 21, 2025

vsethi09 Jan 22, 2025

tivv Jan 21, 2025

sonarqubecloud bot commented Jan 22, 2025

[CDAP-21096] Remove in-memory launching queue from RunRecordMonitorService #15800

Are you sure you want to change the base?

[CDAP-21096] Remove in-memory launching queue from RunRecordMonitorService #15800

Conversation

vsethi09 commented Jan 13, 2025 • edited Loading

Context

Testing

Choose a reason for hiding this comment

vsethi09 Jan 16, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sonarqubecloud bot commented Jan 22, 2025

Quality Gate failed

vsethi09 commented Jan 13, 2025 •

edited

Loading

vsethi09 Jan 16, 2025 •

edited

Loading