chore(blockbuilder): cleanup #15730
chore(blockbuilder): cleanup #15730
Conversation
owen-d
commented
Jan 13, 2025
•
owen-d
commented
- Rewrites queue for more reliable & safe state transitions (no more negative pending jobs)
- Handles off-by-1 polling error where builders would poll indefinitely for the final offset of a partition which didn't exist
- Adds backoff & error propagation logic for polling so jobs can fail after 3 unsuccessful attempts.
fails job after 3 successive kafka polling errors
owen-d
changed the title
pkg/blockbuilder/builder/builder.go
Outdated
| @@ -592,7 +616,7 @@ func (i *BlockBuilder) loadRecords(ctx context.Context, c *kgo.Client, partition | |||
| } | |||
| } | |||
|
|
|||
| return lastConsumedOffset, boff.Err() | |||
| return lastSeenOffset, boff.Err() | |||
There was a problem hiding this comment.
Yeah that's a fair point. The reason I removed this was because outside of logging, it's not used anywhere. It's returned up the chain and logged, but it doesn't influence behavior. I figured reducing unnecessary complexity made sense in this case. The new behavior is we're logging the last seen offset, rather than the last consumed offset. However, I should probably just log the final consumed offset (rather than seen), but not return thread it through our code unnecessarily.
| startOffset := max(partitionOffset.Commit.At+1, partitionOffset.Start.Offset) | ||
| // Likewise, endOffset is initially the next available offset: this is why we treat jobs as end-exclusive: |
There was a problem hiding this comment.
shouldn't we include the endOffset since it is a valid offset for a record in the partition? we'd be skipping the last record otherwise
There was a problem hiding this comment.
WDYM? end offset in this case is exclusive. This is the next available record in partitions which we're caught up on. Trying to poll for these when the partition is no longer being written to causes us to hang forever waiting for data which will not appear. This is the reason we were seeing infinite polling on defunct partitions.
pkg/blockbuilder/scheduler/queue.go
Outdated
| @@ -88,25 +88,21 @@ type JobQueue struct { | |||
| metrics *jobQueueMetrics | |||
| } | |||
|
|
|||
| // NewJobQueue creates a new job queue instance | |||
| // NewJobQueue creates a new JobQueue2 instance | |||
There was a problem hiding this comment.
nit: needs correction
| q.metrics.inProgress.Inc() | ||
| job.StartTime = job.UpdateTime | ||
| case types.JobStatusComplete, types.JobStatusFailed, types.JobStatusExpired: | ||
| q.completed.Push(job) |
There was a problem hiding this comment.
nit: we might have to remove the evicted job from status map
pkg/blockbuilder/scheduler/queue.go
Outdated
| q.statusMap[jobMeta.ID()] = types.JobStatusInProgress | ||
| q.metrics.inProgress.Inc() | ||
| if finished { | ||
| level.Debug(q.logger).Log("msg", "ignoring transition for completed job; will recreate", "id", jobID, "from", currentStatus, "to", to) |
There was a problem hiding this comment.
is this comment correct?
There was a problem hiding this comment.
Yes, but perhaps I can be clearer: we're not going to delete the old completed entry for this job. Instead we're going to recreate a copy.
| beforeSync := time.Now() | ||
| q.SyncJob(jobID, job) | ||
| afterSync := time.Now() | ||
| func TestJobQueue2_TransitionState(t *testing.T) { |
There was a problem hiding this comment.
rename tests that contain Queue2
