lower the scan interval

eph/scheduler.go

@@ -39,7 +39,7 @@ var (
 
 const (
 	// DefaultLeaseRenewalInterval defines the default amount of time between lease renewal attempts
-	DefaultLeaseRenewalInterval = 20 * time.Second
+	DefaultLeaseRenewalInterval = 10 * time.Second
 
 	// DefaultLeaseDuration defines the default amount of time a lease is valid
 	DefaultLeaseDuration = 60 * time.Second

storage/eph_test.go

@@ -97,8 +97,9 @@ func (ts *testSuite) TestMultiple() {
 
 		processor, err := ts.newStorageBackedEPHOptions(*hub.Name, leaserCheckpointer, leaserCheckpointer)
 		ts.Require().NoError(err)
+
 		processors[processor.GetName()] = processor
-		processor.StartNonBlocking(ctx)
+		ts.Require().NoError(processor.StartNonBlocking(ctx))
 		processorNames[i] = processor.GetName()
 	}
 
@@ -319,7 +320,18 @@ func allHandled(partitionsByProcessor map[string][]int, numberOfPartitions int)
 		}
 	}
 
-	fmt.Println(partitionsByProcessor)
+	//var keys []string
+	//for key := range partitionsByProcessor {
+	//	keys = append(keys, key)
+	//}
+	//sort.Strings(keys)
+	//for _, key := range keys {
+	//	ints := partitionsByProcessor[key]
+	//	sort.Ints(ints)
+	//	fmt.Printf("Processor: %q, Partitions %+v\n", key, ints)
+	//}
+	//fmt.Println("========================================")
+	//fmt.Println("========================================")
 
 	for _, count := range countByPartition {
 		if count != 1 {

storage/storage.go

@@ -330,6 +330,7 @@ func (sl *LeaserCheckpointer) ReleaseLease(ctx context.Context, partitionID stri
 		return false, err
 	}
 	delete(sl.leases, partitionID)
+	delete(sl.dirtyPartitions, partitionID)
 	return true, nil
 }
 
@@ -511,9 +512,8 @@ func (sl *LeaserCheckpointer) persistDirtyPartitions(ctx context.Context) error
 			if res.Err != nil {
 				fmt.Printf("run: %d, err: %+v\n", i, res.Err)
 				lastErr = res.Err
-			} else {
-				delete(sl.dirtyPartitions, res.PartitionID)
 			}
+			delete(sl.dirtyPartitions, res.PartitionID)
 		}
 	}
 	return lastErr