Fixed issue that shuffle sharding planner return error if block is under visit by other compactor (#5188)

* Updated visit marker check logic in shuffle sharding planner to not return error if block is under visit by other compactor

Signed-off-by: Alex Le <[email protected]>

* Update changelog

Signed-off-by: Alex Le <[email protected]>

* trigger workflow

Signed-off-by: Alex Le <[email protected]>

* trigger workflow

Signed-off-by: Alex Le <[email protected]>

---------

Signed-off-by: Alex Le <[email protected]>

Author: alexqyle

CHANGELOG.md

@@ -45,6 +45,7 @@
 * [BUGFIX] Ring: Fix case when dynamodb kv reaches the limit of 25 actions per batch call. #5136
 * [BUGFIX] Query-frontend: Fix shardable instant queries do not produce sorted results for `sort`, `sort_desc`, `topk`, `bottomk` functions. #5148, #5170
 * [BUGFIX] Querier: Fix `/api/v1/series` returning 5XX instead of 4XX when limits are hit. #5169
+* [BUGFIX] Compactor: Fix issue that shuffle sharding planner return error if block is under visit by other compactor. #5188
 * [FEATURE] Alertmanager: Add support for time_intervals. #5102
 
 ## 1.14.0 2022-12-02

pkg/compactor/shuffle_sharding_planner.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
 	"github.com/oklog/ulid"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/thanos-io/objstore"
@@ -79,7 +80,8 @@ func (p *ShuffleShardingPlanner) Plan(_ context.Context, metasByMinTime []*metad
 			return nil, fmt.Errorf("unable to get visit marker file for block %s: %s", blockID, err.Error())
 		}
 		if !blockVisitMarker.isVisitedByCompactor(p.blockVisitMarkerTimeout, p.ringLifecyclerID) {
-			return nil, fmt.Errorf("block %s is not visited by current compactor %s", blockID, p.ringLifecyclerID)
+			level.Warn(p.logger).Log("msg", "block is not visited by current compactor", "block_id", blockID, "compactor_id", p.ringLifecyclerID)
+			return nil, nil
 		}
 
 		resultMetas = append(resultMetas, b)

pkg/compactor/shuffle_sharding_planner_test.go

@@ -306,7 +306,7 @@ func TestShuffleShardingPlanner_Plan(t *testing.T) {
 					compactorID: otherCompactor,
 				},
 			},
-			expectedErr: fmt.Errorf("block %s is not visited by current compactor %s", block1ulid.String(), currentCompactor),
+			expected: []*metadata.Meta{},
 		},
 		"test should not compact if visit marker file is expired": {
 			ranges: []int64{2 * time.Hour.Milliseconds()},
@@ -333,7 +333,7 @@ func TestShuffleShardingPlanner_Plan(t *testing.T) {
 					compactorID: currentCompactor,
 				},
 			},
-			expectedErr: fmt.Errorf("block %s is not visited by current compactor %s", block1ulid.String(), currentCompactor),
+			expected: []*metadata.Meta{},
 		},
 	}