This is an automated email from the ASF dual-hosted git repository. damondouglas pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push: new 5b44d3637c8 [Playground] derf Stop tickers to avoid leaks. (#26507) 5b44d3637c8 is described below commit 5b44d3637c82d8844515e701fa14a065d7849015 Author: Robert Burke <lostl...@users.noreply.github.com> AuthorDate: Wed May 3 16:39:43 2023 -0700 [Playground] derf Stop tickers to avoid leaks. (#26507) Co-authored-by: lostluck <13907733+lostl...@users.noreply.github.com> --- playground/backend/internal/cache/local/local_cache.go | 1 + playground/backend/internal/code_processing/code_processing.go | 6 +++--- playground/backend/internal/db/datastore/emulator_wrapper.go | 2 ++ playground/backend/internal/emulators/kafka.go | 7 ++++--- 4 files changed, 10 insertions(+), 6 deletions(-) diff --git a/playground/backend/internal/cache/local/local_cache.go b/playground/backend/internal/cache/local/local_cache.go index 72568540710..881b6b523d2 100644 --- a/playground/backend/internal/cache/local/local_cache.go +++ b/playground/backend/internal/cache/local/local_cache.go @@ -166,6 +166,7 @@ func (lc *Cache) GetSdkCatalog(_ context.Context) ([]*entity.SDKEntity, error) { func (lc *Cache) startGC(ctx context.Context) { ticker := time.NewTicker(lc.cleanupInterval) + defer ticker.Stop() for { select { case <-ctx.Done(): diff --git a/playground/backend/internal/code_processing/code_processing.go b/playground/backend/internal/code_processing/code_processing.go index be4c88709ba..d371860430e 100644 --- a/playground/backend/internal/code_processing/code_processing.go +++ b/playground/backend/internal/code_processing/code_processing.go @@ -434,10 +434,10 @@ func reconcileBackgroundTask(pipelineLifeCycleCtx context.Context, pipelineId uu // If cancel flag exists, and it is true it means that the code processing was canceled. Set true to cancelChannel and return. func cancelCheck(ctx context.Context, pipelineId uuid.UUID, cancelFunc context.CancelFunc, cacheService cache.Cache) { ticker := time.NewTicker(pauseDuration) + defer ticker.Stop() for { select { case <-ctx.Done(): - ticker.Stop() return case <-ticker.C: // Use background context for the cache operation to avoid failure when the main context is timed out. @@ -460,12 +460,12 @@ func cancelCheck(ctx context.Context, pipelineId uuid.UUID, cancelFunc context.C // In other case each pauseDuration checks that graph file exists or not and try to save it to the cache. func readGraphFile(pipelineLifeCycleCtx context.Context, cacheService cache.Cache, graphFilePath string, pipelineId uuid.UUID) { ticker := time.NewTicker(pauseDuration) + defer ticker.Stop() for { select { // waiting when graph file appears case <-ticker.C: if _, err := os.Stat(graphFilePath); err == nil { - ticker.Stop() graph, err := os.ReadFile(graphFilePath) if err != nil { logger.Errorf("%s: Error during saving graph to the file: %s", pipelineId, err.Error()) @@ -474,7 +474,6 @@ func readGraphFile(pipelineLifeCycleCtx context.Context, cacheService cache.Cach } // in case of timeout or cancel case <-pipelineLifeCycleCtx.Done(): - ticker.Stop() if _, err := os.Stat(graphFilePath); err == nil { graph, err := os.ReadFile(graphFilePath) if err != nil { @@ -497,6 +496,7 @@ func readGraphFile(pipelineLifeCycleCtx context.Context, cacheService cache.Cach // In other case each pauseDuration write to cache logs of the code processing. func readLogFile(pipelineLifeCycleCtx context.Context, cacheService cache.Cache, logFilePath string, pipelineId uuid.UUID, stopReadLogsChannel, finishReadLogChannel chan bool) { ticker := time.NewTicker(pauseDuration) + defer ticker.Stop() for { select { // in case of timeout or cancel diff --git a/playground/backend/internal/db/datastore/emulator_wrapper.go b/playground/backend/internal/db/datastore/emulator_wrapper.go index 9252c454b01..f1cf0206b66 100644 --- a/playground/backend/internal/db/datastore/emulator_wrapper.go +++ b/playground/backend/internal/db/datastore/emulator_wrapper.go @@ -130,7 +130,9 @@ func startEmulator() (*emulator, error) { ok := func() bool { workTicker := time.NewTicker(pauseDuration) + defer workTicker.Stop() globalTicker := time.NewTicker(waitDuration) + defer workTicker.Stop() for { select { case <-workTicker.C: diff --git a/playground/backend/internal/emulators/kafka.go b/playground/backend/internal/emulators/kafka.go index 58112439c67..031373b47b0 100644 --- a/playground/backend/internal/emulators/kafka.go +++ b/playground/backend/internal/emulators/kafka.go @@ -16,7 +16,6 @@ package emulators import ( - "beam.apache.org/playground/backend/internal/logger" "bufio" "encoding/json" "errors" @@ -28,10 +27,10 @@ import ( "strings" "time" + "beam.apache.org/playground/backend/internal/constants" + "beam.apache.org/playground/backend/internal/logger" "github.com/confluentinc/confluent-kafka-go/kafka" "github.com/linkedin/goavro" - - "beam.apache.org/playground/backend/internal/constants" ) const ( @@ -109,7 +108,9 @@ func NewKafkaMockCluster(emulatorExecutablePath string) (*KafkaMockCluster, erro bootstrapServers := fmt.Sprintf("%s%s%s", host, addressSeperator, port) workTicker := time.NewTicker(pauseDuration) + defer workTicker.Stop() globalTicker := time.NewTicker(globalDuration) + defer globalTicker.Stop() for { select { case <-workTicker.C: