Allow calling Stop multiple times on RetryWatcher

This makes the Stop method idempotent so that if Stop is called multiple
times, it does not cause a panic due to closing a closed channel.

Signed-off-by: mprahl <mprahl@users.noreply.github.com>

Kubernetes-commit: a54ba917be42c941edf1a0359dced04e1a5e1d6f

tools/watch/retrywatcher.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"sync"
 	"time"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -53,6 +54,7 @@ type RetryWatcher struct {
 	stopChan            chan struct{}
 	doneChan            chan struct{}
 	minRestartDelay     time.Duration
+	stopChanLock        sync.Mutex
 }
 
 // NewRetryWatcher creates a new RetryWatcher.
@@ -286,7 +288,15 @@ func (rw *RetryWatcher) ResultChan() <-chan watch.Event {
 
 // Stop implements Interface.
 func (rw *RetryWatcher) Stop() {
+	rw.stopChanLock.Lock()
+	defer rw.stopChanLock.Unlock()
+
+	// Prevent closing an already closed channel to prevent a panic
+	select {
+	case <-rw.stopChan:
+	default:
 		close(rw.stopChan)
+	}
 }
 
 // Done allows the caller to be notified when Retry watcher stops.

tools/watch/retrywatcher_test.go

@@ -585,6 +585,8 @@ func TestRetryWatcherToFinishWithUnreadEvents(t *testing.T) {
 	// Give the watcher a chance to get to sending events (blocking)
 	time.Sleep(10 * time.Millisecond)
 
+	watcher.Stop()
+	// Verify a second stop does not cause a panic
 	watcher.Stop()
 
 	maxTime := time.Second