[GOBBLIN-2171] Streamline log messages for easier debugging of GaaS multi-active DAG processing

[phet]

Dear Gobblin maintainers,

Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!

JIRA

• My PR addresses the following Gobblin JIRA issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
  • https://issues.apache.org/jira/browse/GOBBLIN-2171

Description

• Here are some details about my PR, including screenshots (if applicable):

Some logs from GaaS flow/DAG processing, e.g. by the MysqlMultiActiveLeaseArbiter, the FlowLaunchHandler, the DagProcessingEngine, etc., are challenging to skim, especially when only the first characters on the line are shown (as the remaining content of long lines flows beyond view).

Improve consistency, brevity, and the placement of the most-vital info. Also contextualize when it's not otherwise clear where in the processing a message originates (e.g. which MultiActiveLeaseArbiter instance).

Tests

• My PR adds the following unit tests OR does not need testing for this extremely good reason:

existing tests still pass

Commits

• My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

[khandelwal-prateek]

Can you please add the log snippet for a test flow execution before & after the change in testing section

[phet]

Can you please add the log snippet for a test flow execution before & after the change in testing section

here are some excerpts:

MysqlMultiActiveLeaseArbiter initialized - mysql_multi_active_lease_arbiter_store
Multi-active arbitration for [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testFlowGroup, flowName=testFlowName, flowExecutionId=12345677, jobName=testJobName, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=1710451837)
Successfully leased [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testFlowGroup, flowName=testFlowName, flowExecutionId=1731481279160, jobName=testJobName, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=1731481279160)
Multi-active lease status [LeaseObtainedStatus] for [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testFlowGroup, flowName=testFlowName, flowExecutionId=1731481279160, jobName=testJobName, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=1731481279160)
Multi-active will use DB time (1731481279204) to launder [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testFlowGroup, flowName=testFlowName, flowExecutionId=12345677, jobName=testJobName, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=1710451837)
Multi-active lease status [LeasedToAnotherStatus] for [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testFlowGroup, flowName=testFlowName, flowExecutionId=1731481279160, jobName=testJobName, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=1731481279160)
Multi-active lease COMPLETED [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testFlowGroup, flowName=testFlowName, flowExecutionId=1731481310422, jobName=testJobName, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=1731481310422) - no longer leasing this event
Multi-active lease status [NoLongerLeasingStatus] for [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testFlowGroup, flowName=testFlowName, flowExecutionId=12345677, jobName=testJobName, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=1710451837)
tryAcquireLease for [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testFlowGroup, flowName=7f59d05a-d, flowExecutionId=12345677, jobName=testJobName, dagActionType=LAUNCH), isReminder=true, eventTimeMillis=1731481321596) (dbEventTimeMillis: 2024-11-12 23:02:01.597) - Newer DB time, so discarding out-of-date reminder

and

Orchestrator - onAdd[Topology]Spec: gobblinExecutor/1
Orchestrator - onAdd[Flow]Spec: gobblin-flow:/testGroup1/testFlow1/
Handling trigger DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testGroup1, flowName=testFlow1, flowExecutionId=1731478778356, jobName=, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=0) (adoptConsensusTimestamp: true)
Multi-active arbitration for [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testGroup1, flowName=testFlow1, flowExecutionId=1731478778356, jobName=, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=0)
Successfully leased [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testGroup1, flowName=testFlow1, flowExecutionId=1731478778474, jobName=, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=1731478778474)
Multi-active lease status [LeaseObtainedStatus] for [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testGroup1, flowName=testFlow1, flowExecutionId=1731478778474, jobName=, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=1731478778474)
Adding Dag Action for flowGroup testGroup1, flowName testFlow1, flowExecutionId 1731478778474, jobName , dagActionType LAUNCH
Multi-active lease COMPLETED [mysql_multi_active_lease_arbiter_store] DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testGroup1, flowName=testFlow1, flowExecutionId=1731478778474, jobName=, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=1731478778474) - no longer leasing this event
Successfully persisted (origEventTimestamp: 0) DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testGroup1, flowName=testFlow1, flowExecutionId=1731478778474, jobName=, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=1731478778474)
Multi-active scheduler finished handling DagActionStore.LeaseParams(dagAction=DagActionStore.DagAction(flowGroup=testGroup1, flowName=testFlow1, flowExecutionId=1731478778356, jobName=, dagActionType=LAUNCH), isReminder=false, eventTimeMillis=0)

and

MockedDagProc - initializing - dagId : fg-3_fn-3_1237
MockedDagProc - ready to process - dagId : fg-3_fn-3_1237
MockedDagProc - processed - dagId : fg-3_fn-3_1237
Deleting DagActionStore.DagAction(flowGroup=fg-1, flowName=fn-1, flowExecutionId=1235, jobName=jn-1, dagActionType=LAUNCH)
MockedDagProc - concluded dagTask - dagId : fg-1_fn-1_1235