Merge pull request #2269 from zerowidth/fix-admin-describe-log-dir-hang

Fix DescribeLogDirs hang in admin client

admin.go

@@ -1050,12 +1050,12 @@ func (ca *clusterAdmin) DescribeLogDirs(brokerIds []int32) (allLogDirs map[int32
 	wg := sync.WaitGroup{}
 
 	for _, b := range brokerIds {
-		wg.Add(1)
 		broker, err := ca.findBroker(b)
 		if err != nil {
 			Logger.Printf("Unable to find broker with ID = %v\n", b)
 			continue
 		}
+		wg.Add(1)
 		go func(b *Broker, conf *Config) {
 			defer wg.Done()
 			_ = b.Open(conf) // Ensure that broker is opened

admin_test.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"
 )
 
 func TestClusterAdmin(t *testing.T) {
@@ -1742,3 +1743,48 @@ func TestDescribeLogDirs(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestDescribeLogDirsUnknownBroker(t *testing.T) {
+	seedBroker := NewMockBroker(t, 1)
+	defer seedBroker.Close()
+
+	seedBroker.SetHandlerByMap(map[string]MockResponse{
+		"MetadataRequest": NewMockMetadataResponse(t).
+			SetController(seedBroker.BrokerID()).
+			SetBroker(seedBroker.Addr(), seedBroker.BrokerID()),
+		"DescribeLogDirsRequest": NewMockDescribeLogDirsResponse(t).
+			SetLogDirs("/tmp/logs", map[string]int{"topic1": 2, "topic2": 2}),
+	})
+
+	config := NewTestConfig()
+	config.Version = V1_0_0_0
+
+	admin, err := NewClusterAdmin([]string{seedBroker.Addr()}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	type result struct {
+		metadata map[int32][]DescribeLogDirsResponseDirMetadata
+		err      error
+	}
+
+	res := make(chan result)
+
+	go func() {
+		metadata, err := admin.DescribeLogDirs([]int32{seedBroker.BrokerID() + 1})
+		res <- result{metadata, err}
+	}()
+
+	select {
+	case <-time.After(time.Second):
+		t.Fatalf("DescribeLogDirs timed out")
+	case returned := <-res:
+		if len(returned.metadata) != 0 {
+			t.Fatalf("Expected no results, got %v", len(returned.metadata))
+		}
+		if returned.err != nil {
+			t.Fatalf("Expected no error, got %v", returned.err)
+		}
+	}
+}