[CDAP-21096] Remove in-memory launching queue from RunRecordMonitorService #15800
Conversation
vsethi09
force-pushed
the
feature/CDAP-21096_fix_RunRecordMonitorService_queue_cache
branch
2 times, most recently
from
ef61989 to
00add6d
Compare
|
vsethi09
force-pushed
the
feature/CDAP-21096_fix_RunRecordMonitorService_queue_cache
branch
3 times, most recently
from
dc068c3 to
0ea30c6
Compare
vsethi09
changed the title
vsethi09
force-pushed
the
feature/CDAP-21096_fix_RunRecordMonitorService_queue_cache
branch
from
0ea30c6 to
abcd4c5
Compare
| * @return Counter with total number of launching and running program runs. | ||
| */ | ||
| public Counter getCount() { | ||
| return getFlowControlMetrics(true, true); |
There was a problem hiding this comment.
Let's not use term metrics as it's pretty strongly associated with metrics service. Use counts.
There was a problem hiding this comment.
Renamed with Counter.
| } | ||
|
|
||
| int launchingCount = addRequest(programRunId, programOptions, programDescriptor); | ||
| int runningCount = getFlowControlMetrics(false, true).getRunningCount(); |
There was a problem hiding this comment.
Are there value in having getFlowControlMetrics method? in addRequestAndGetCount it still ends up creating 2 separate database transactions. I'd consider moving transaction into addRequestAndGetCount/getCount and removing/splitting getFlowControlMetrics into 2 methods without transaction handling.
As of emitFlowControlMetrics, as far as I can see in all but edge case all metrics are emitted, so I'd just make it emit both metrics universaly.
| if (runRecordDetail.getStatus() == ProgramRunStatus.PENDING) { | ||
| runRecordMonitorService.addRequest(runRecordDetail.getProgramRunId()); | ||
| flowControlMonitorService.addRequest(runRecordDetail.getProgramRunId(), |
There was a problem hiding this comment.
I don't think this is needed if the run is already in the database.
There was a problem hiding this comment.
This is required for emitting flow control metrics.
Replaces addRequest() with emitFlowControlMetrics() call.
There was a problem hiding this comment.
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?
| throws IOException { | ||
| long startTs = RunIds.getTime(programRunId.getRun(), TimeUnit.SECONDS); | ||
| if (startTs == -1L) { | ||
| LOG.error( |
There was a problem hiding this comment.
I would fail it. Since we are in the handler, we can fail and pass it to user. We could not do it in the subscriber
There was a problem hiding this comment.
Done.
Throwing IllegalArgumentException for this case.
| "Ignoring unexpected request to record rejected state for program run {} that has an existing " | ||
| + "run record in run state {} and cluster state {}.", | ||
| programRunId, existing.getStatus(), existing.getCluster().getStatus()); | ||
| "Ignoring unexpected request to record rejected state for program run {} that has no existing run record.", |
There was a problem hiding this comment.
Why? I don't know in which case it may happen, but I'd still leave a trace. Just skip delete.
| */ | ||
| public int getLaunchingCount(Set<ProgramType> programTypes, @Nullable Integer limit) throws IOException { | ||
| AtomicInteger count = new AtomicInteger(0); | ||
| try (CloseableIterator<RunRecordDetail> iterator = queryProgramRuns( |
There was a problem hiding this comment.
This is heavy. Can we use io.cdap.cdap.spi.data.StructuredTable#count? If needed, we can even add a field when we write the record to filter efficiently
vsethi09
force-pushed
the
feature/CDAP-21096_fix_RunRecordMonitorService_queue_cache
branch
from
abcd4c5 to
343101b
Compare
Context
Multiple instances of
RunRecordMonitoringServicecannot 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
AppMetadataStoreAPIs.For more context, see: #15773 (comment).
Note:
RunRecordMonitoringServiceis renamed asFlowControlMonitoringService.Testing