IG-19366: if the shard does not exist yet, turn seek-latest to seek-earliest

Author: assaf758

pkg/dataplane/streamconsumergroup/claim.go

@@ -222,22 +222,30 @@ func (c *claim) fetchRecordBatch(location string) (string, error) {
 
 func (c *claim) getCurrentShardLocation(shardID int) (string, error) {
 
+	initialLocation := c.member.streamConsumerGroup.config.Claim.RecordBatchFetch.InitialLocation
+
 	// get the location from persistency
-	currentShardLocation, err := c.member.streamConsumerGroup.getShardLocationFromPersistency(shardID)
+	currentShardLocation, err := c.member.streamConsumerGroup.getShardLocationFromPersistency(shardID, initialLocation)
 	if err != nil && errors.RootCause(err) != ErrShardNotFound {
 		return "", errors.Wrap(err, "Failed to get shard location from persistency")
 	}
 
 	// if shard wasn't found, try again periodically
 	if errors.RootCause(err) == ErrShardNotFound {
+
+		// if the shard does not exist yet, turn seek-latest to seek-earliest (IG-19366)
+		if initialLocation == v3io.SeekShardInputTypeLatest {
+			initialLocation = v3io.SeekShardInputTypeEarliest
+		}
+
 		for {
 			select {
 
 			// TODO: from configuration
 			case <-time.After(c.member.streamConsumerGroup.config.SequenceNumber.ShardWaitInterval):
 
 				// get the location from persistency
-				currentShardLocation, err = c.member.streamConsumerGroup.getShardLocationFromPersistency(shardID)
+				currentShardLocation, err = c.member.streamConsumerGroup.getShardLocationFromPersistency(shardID, initialLocation)
 				if err != nil {
 					if errors.RootCause(err) == ErrShardNotFound {
 

pkg/dataplane/streamconsumergroup/streamconsumergroup.go

@@ -208,7 +208,7 @@ func (scg *streamConsumerGroup) getStateFilePath() string {
 	return path.Join(scg.streamPath, fmt.Sprintf("%s-state.json", scg.name))
 }
 
-func (scg *streamConsumerGroup) getShardLocationFromPersistency(shardID int) (string, error) {
+func (scg *streamConsumerGroup) getShardLocationFromPersistency(shardID int, initialLocation v3io.SeekShardInputType) (string, error) {
 	scg.logger.DebugWith("Getting shard sequenceNumber from persistency", "shardID", shardID)
 
 	seekShardInput := v3io.SeekShardInput{}
@@ -223,7 +223,7 @@ func (scg *streamConsumerGroup) getShardLocationFromPersistency(shardID int) (st
 			return "", errors.Wrap(err, "Failed to get shard sequenceNumber from item attributes")
 		}
 
-		seekShardInput.Type = scg.config.Claim.RecordBatchFetch.InitialLocation
+		seekShardInput.Type = initialLocation
 	} else {
 
 		// use sequence number