Add additional sync timing information

[evantahler]

Closes #17407

whimsical link

This PR adds a plethora of additional time information for each part of the replication workflow so we can better analyze performance.

---

The times that are added are different from the existing startTime and endTime in subtle but interesting ways:

• startTime - the start of the workflow
• replication_start_time - the start of the replication activity
• replication_end_time - the end of the replication activity
• source_read_start_time - the start of the source container
• source_read_end_time - the end of the source container
• destination_write_start_time - the start of the destination container
• destination_write_end_time - the end of the destination container
• normalization_start_time - the start of the normalization container
• normalization_end_time - the end of the normalization container
• endTime - the conclusion of the workflow

While it might appear that startTime, replication_start_time, and source_read_start_time should all be the same, there is often a lag, which is due to things like downloading the docker image, starting the K8s pod, etc.

---

{
  "status" : "completed",
  "recordsSynced" : 1222,
  "bytesSynced" : 344396,
  "startTime" : 1665531403219, // Exists
  "endTime" : 1665531405935, // Exists
  "totalStats" : {
    "recordsEmitted" : 1222,
    "bytesEmitted" : 344396,
    "sourceStateMessagesEmitted" : 8,
    "destinationStateMessagesEmitted" : 1,
    "recordsCommitted" : 1222,
    "meanSecondsBeforeSourceStateMessageEmitted" : 0,
    "maxSecondsBeforeSourceStateMessageEmitted" : 0,
    "maxSecondsBetweenStateMessageEmittedandCommitted" : 1,
    "meanSecondsBetweenStateMessageEmittedandCommitted" : 1,
    "replicationStartTime" : 1665531403219, // NEW!
    "replicationEndTime" : 1665531405932,  // NEW!
    "sourceReadStartTime" : 1665531403274, // NEW!
    "sourceReadEndTime" : 1665531405623, // NEW!
    "destinationWriteStartTime" : 1665531403339, // NEW!
    "destinationWriteEndTime" : 1665531405931 // NEW!
  },
}

Confirming that the data appears in BiqQuery:

select
  attempt_completion_status
  , attempt_stage
  , connection_id
  , workspace_id
  , connector_source
  , connector_destination
  , job_type
  , replication_start_time
  , replication_end_time
  , source_read_start_time
  , source_read_end_time
  , destination_write_start_time
  , destination_write_end_time
  , normalization_start_time
  , normalization_end_time
from api.connector_jobs 
WHERE 1=1
AND loaded_at >= "2022-09-01 00:00:00"
AND workspace_id = '40c15f0a-027e-48ef-943f-956e1258264a' 
order by loaded_at desc
limit 100

Produces

attempt_completion_status	attempt_stage	connection_id	workspace_id	connector_source	connector_destination	job_type	replication_start_time	replication_end_time	source_read_start_time	source_read_end_time	destination_write_start_time	destination_write_end_time	normalization_start_time	normalization_end_time
SUCCEEDED	ENDED	335c7727-42aa-4331-8846-7b33348de803	40c15f0a-027e-48ef-943f-956e1258264a	Faker	Postgres	SYNC	1665442059649	1665442061312	1665442059700	1665442060972	1665442059753	1665442061311	1665442061493	1665442115182

[evantahler]

It seems that these longs needed to be initialized or else they couldn't be used below in withSourceReadStartTime() - I'll be there's an Optional<> Java thing I don't know about that's better than all the ternaries...

[evantahler]

Tests are finally green! Now ready for review!

[pedroslopez]

LGTM, not too familiar with the whole java threaded synchronization stuff though

[davinchia]

Very nice!

Most of my comments are suggestions/nits for improving readability.

Approving since I don't think these are blocking comments.

[alex-gron]

The names look good to me and make sense. Thanks @evantahler !