Repository: armadaproject/armada
Pull request #2147 (2023-02-20 22:10:37)

Fix partition marker flow when pulsar topics have many partitions
 - Initialise groupId correctly on partition markers - currently all getting the id of 0
 - Merging InsertPartitionMarker is not working as expected - as it doesn't mutate the InsertPartitionMarker at all (it mutates a copy)
    - Fixed by enforcing it is used as a pointer
 
Now all the PartitionMarkers should get populated in the scheduler db with a non-zero group id 


codecov-commenter: # [Codecov](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) Report
Base: **46.37**% // Head: **46.81**% // Increases project coverage by **`+0.43%`** :tada:
> Coverage data is based on head [(`7d08e6e`)](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) compared to base [(`f72e7fb`)](https://codecov.io/gh/armadaproject/armada/commit/f72e7fbda06be88b00952da3217ca88b67eb16e4?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject).
> Patch coverage: 47.65% of modified lines in pull request are covered.

:mega: This organization is not using Codecov’s [GitHub App Integration](https://github.com/apps/codecov). We recommend you install it so Codecov can continue to function properly for your repositories. [Learn more](https://about.codecov.io/blog/codecov-is-updating-its-github-integration/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject)

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #2147      +/-   ##
==========================================
+ Coverage   46.37%   46.81%   +0.43%     
==========================================
  Files         220      222       +2     
  Lines       32228    32486     +258     
==========================================
+ Hits        14945    15207     +262     
+ Misses      16096    16067      -29     
- Partials     1187     1212      +25     
```

| Flag | Coverage Δ | |
|---|---|---|
| armada-server | `46.81% <47.65%> (+0.43%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) | Coverage Δ | |
|---|---|---|
| [internal/armada/server/lease.go](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvYXJtYWRhL3NlcnZlci9sZWFzZS5nbw==) | `8.50% <0.00%> (-0.15%)` | :arrow_down: |
| [internal/executor/application.go](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvZXhlY3V0b3IvYXBwbGljYXRpb24uZ28=) | `9.90% <0.00%> (-0.20%)` | :arrow_down: |
| [internal/executor/job/submit.go](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvZXhlY3V0b3Ivam9iL3N1Ym1pdC5nbw==) | `21.00% <0.00%> (ø)` | |
| [internal/executor/service/cluster\_allocation.go](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvZXhlY3V0b3Ivc2VydmljZS9jbHVzdGVyX2FsbG9jYXRpb24uZ28=) | `0.00% <0.00%> (ø)` | |
| [internal/executor/util/uuid.go](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvZXhlY3V0b3IvdXRpbC91dWlkLmdv) | `0.00% <ø> (ø)` | |
| [internal/scheduler/legacyscheduler.go](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvc2NoZWR1bGVyL2xlZ2FjeXNjaGVkdWxlci5nbw==) | `69.33% <0.00%> (-0.34%)` | :arrow_down: |
| [internal/executor/job/util.go](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvZXhlY3V0b3Ivam9iL3V0aWwuZ28=) | `29.48% <51.11%> (+29.48%)` | :arrow_up: |
| [internal/executor/job/job\_run\_state\_store.go](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvZXhlY3V0b3Ivam9iL2pvYl9ydW5fc3RhdGVfc3RvcmUuZ28=) | `81.13% <81.13%> (ø)` | |
| [internal/common/slices/slices.go](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvY29tbW9uL3NsaWNlcy9zbGljZXMuZ28=) | `97.29% <100.00%> (+0.68%)` | :arrow_up: |
| [internal/common/util/list.go](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject#diff-aW50ZXJuYWwvY29tbW9uL3V0aWwvbGlzdC5nbw==) | `44.68% <100.00%> (+3.77%)` | :arrow_up: |
| ... and [13 more](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/armadaproject/armada/pull/2147?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=armadaproject).


Patch:
diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
index 55e79147..742fd5c4 100644
--- a/internal/scheduler/scheduler.go
+++ b/internal/scheduler/scheduler.go
@@ -599,7 +599,7 @@ func (s *Scheduler) initialise(ctx context.Context) error {
 // function was called. This is achieved firstly by publishing messages to Pulsar and then polling the
 // database until all messages have been written.
 func (s *Scheduler) ensureDbUpToDate(ctx context.Context, pollInterval time.Duration) error {
-	var groupId uuid.UUID
+	groupId := uuid.New()
 	var numSent uint32
 	var err error
 
diff --git a/internal/scheduleringester/dbops.go b/internal/scheduleringester/dbops.go
index 2461d19e..ad4f6825 100644
--- a/internal/scheduleringester/dbops.go
+++ b/internal/scheduleringester/dbops.go
@@ -172,9 +172,9 @@ func (a InsertJobRunErrors) Merge(b DbOperation) bool {
 	return mergeInMap(a, b)
 }
 
-func (a InsertPartitionMarker) Merge(b DbOperation) bool {
+func (a *InsertPartitionMarker) Merge(b DbOperation) bool {
 	switch op := b.(type) {
-	case InsertPartitionMarker:
+	case *InsertPartitionMarker:
 		a.markers = append(a.markers, op.markers...)
 		return true
 	}
@@ -271,7 +271,7 @@ func (a MarkRunsRunning) CanBeAppliedBefore(b DbOperation) bool {
 	return !definesRun(a, b)
 }
 
-func (a InsertPartitionMarker) CanBeAppliedBefore(b DbOperation) bool {
+func (a *InsertPartitionMarker) CanBeAppliedBefore(b DbOperation) bool {
 	// Partition markers can never be brought forward
 	return false
 }
diff --git a/internal/scheduleringester/instructions.go b/internal/scheduleringester/instructions.go
index f4bd5ade..68eadacb 100644
--- a/internal/scheduleringester/instructions.go
+++ b/internal/scheduleringester/instructions.go
@@ -281,7 +281,7 @@ func (c *InstructionConverter) handleCancelledJob(cancelledJob *armadaevents.Can
 }
 
 func (c *InstructionConverter) handlePartitionMarker(pm *armadaevents.PartitionMarker, created time.Time) ([]DbOperation, error) {
-	return []DbOperation{InsertPartitionMarker{
+	return []DbOperation{&InsertPartitionMarker{
 		markers: []*schedulerdb.Marker{
 			{
 				GroupID:     armadaevents.UuidFromProtoUuid(pm.GroupId),
diff --git a/internal/scheduleringester/schedulerdb.go b/internal/scheduleringester/schedulerdb.go
index acbe2169..a2230e76 100644
--- a/internal/scheduleringester/schedulerdb.go
+++ b/internal/scheduleringester/schedulerdb.go
@@ -159,7 +159,7 @@ func (s *SchedulerDb) WriteDbOp(ctx context.Context, op DbOperation) error {
 			i++
 		}
 		return database.Upsert(ctx, s.db, "job_run_errors", records)
-	case InsertPartitionMarker:
+	case *InsertPartitionMarker:
 		for _, marker := range o.markers {
 			err := queries.InsertMarker(ctx, schedulerdb.InsertMarkerParams{
 				GroupID:     marker.GroupID,


Test patch:
diff --git a/internal/scheduleringester/dbops_test.go b/internal/scheduleringester/dbops_test.go
index 7a219509..de1e984a 100644
--- a/internal/scheduleringester/dbops_test.go
+++ b/internal/scheduleringester/dbops_test.go
@@ -29,6 +29,31 @@ func TestMerge(t *testing.T) {
 	assert.Equal(t, MarkJobsCancelRequested{jobId1: false, jobId2: true, jobId3: true}, markJobsCancelled1)
 }
 
+func TestMerge_InsertPartitionMarker(t *testing.T) {
+	marker1 := &InsertPartitionMarker{markers: []*schedulerdb.Marker{
+		{
+			PartitionID: int32(1),
+		},
+	}}
+	marker2 := &InsertPartitionMarker{markers: []*schedulerdb.Marker{
+		{
+			PartitionID: int32(2),
+		},
+	}}
+	expectedOutput := &InsertPartitionMarker{markers: []*schedulerdb.Marker{
+		{
+			PartitionID: int32(1),
+		},
+		{
+			PartitionID: int32(2),
+		},
+	}}
+
+	marker1.Merge(marker2)
+
+	assert.Equal(t, expectedOutput, marker1)
+}
+
 // Test that db op optimisation
 // 1. produces the expected number of ops after optimisations and
 // 2. results in the same end state as if no optimisation had been applied.
@@ -142,7 +167,7 @@ func TestDbOperationOptimisation(t *testing.T) {
 		}},
 		"InsertPartitionMarker": {N: 2, Ops: []DbOperation{
 			InsertJobs{jobIds[0]: &schedulerdb.Job{JobID: jobIds[0]}}, // 1
-			InsertPartitionMarker{markers: []*schedulerdb.Marker{}},   // 2
+			&InsertPartitionMarker{markers: []*schedulerdb.Marker{}},  // 2
 			InsertJobs{jobIds[1]: &schedulerdb.Job{JobID: jobIds[1]}}, // 1
 			InsertJobs{jobIds[2]: &schedulerdb.Job{JobID: jobIds[2]}}, // 1
 		}},
diff --git a/internal/scheduleringester/instructions_test.go b/internal/scheduleringester/instructions_test.go
index bd8f2381..6e01b72d 100644
--- a/internal/scheduleringester/instructions_test.go
+++ b/internal/scheduleringester/instructions_test.go
@@ -153,7 +153,7 @@ func TestConvertSequence(t *testing.T) {
 		"PositionMarker": {
 			events: []*armadaevents.EventSequence_Event{f.PartitionMarker},
 			expected: []DbOperation{
-				InsertPartitionMarker{markers: []*schedulerdb.Marker{
+				&InsertPartitionMarker{markers: []*schedulerdb.Marker{
 					{
 						GroupID:     f.PartitionMarkerGroupIdUuid,
 						PartitionID: f.PartitionMarkerPartitionId,
diff --git a/internal/scheduleringester/schedulerdb_test.go b/internal/scheduleringester/schedulerdb_test.go
index 79553b6d..47a76194 100644
--- a/internal/scheduleringester/schedulerdb_test.go
+++ b/internal/scheduleringester/schedulerdb_test.go
@@ -202,7 +202,7 @@ func TestWriteOps(t *testing.T) {
 			},
 		}},
 		"Insert PositionMarkers": {Ops: []DbOperation{
-			InsertPartitionMarker{
+			&InsertPartitionMarker{
 				markers: []*schedulerdb.Marker{
 					{
 						GroupID:     uuid.New(),
@@ -509,7 +509,7 @@ func assertOpSuccess(t *testing.T, schedulerDb *SchedulerDb, serials map[string]
 			}
 		}
 		assert.Equal(t, expected, actual)
-	case InsertPartitionMarker:
+	case *InsertPartitionMarker:
 		actual, err := queries.SelectAllMarkers(ctx)
 		require.NoError(t, err)
 		require.Equal(t, len(expected.markers), len(actual))