From 2f76283f03d33c86fd30be3ceeaad78be198b095 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Sat, 28 Dec 2019 12:31:35 -0700 Subject: [PATCH] fix replication errors & test races --- ctl/import_test.go | 29 ++++++++++++++++++----------- holder.go | 17 +++++++++++++++++ server/handler_test.go | 2 +- 3 files changed, 36 insertions(+), 12 deletions(-) diff --git a/ctl/import_test.go b/ctl/import_test.go index 62b9b3c33..5c41a840c 100644 --- a/ctl/import_test.go +++ b/ctl/import_test.go @@ -25,6 +25,7 @@ import ( "reflect" "strings" "testing" + "time" "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/test" @@ -299,20 +300,26 @@ func TestImportCommand_KeyReplication(t *testing.T) { // Verify that the data is available on both nodes. for _, host := range []string{host0, host1} { - qry := "Count(Row(f=foo0))" - resp, err := http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+host+"/index/i/query", strings.NewReader(qry))) - if err != nil { - t.Fatalf("Querying data for validation: %s", err) - } + if err := test.RetryUntil(2*time.Second, func() error { + qry := "Count(Row(f=foo0))" + resp, err := http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+host+"/index/i/query", strings.NewReader(qry))) + if err != nil { + return fmt.Errorf("Querying data for validation: %s", err) + } - // Read body and unmarshal response. - exp := `{"results":[100]}` + "\n" - if body, err := ioutil.ReadAll(resp.Body); err != nil { - t.Fatalf("reading: %s", err) - } else if !reflect.DeepEqual(body, []byte(exp)) { - t.Errorf("expected: %s, but got: %s", exp, body) + // Read body and unmarshal response. + exp := `{"results":[100]}` + "\n" + if body, err := ioutil.ReadAll(resp.Body); err != nil { + return fmt.Errorf("reading: %s", err) + } else if !reflect.DeepEqual(body, []byte(exp)) { + return fmt.Errorf("expected: %s, but got: %s", exp, body) + } + return nil + }); err != nil { + t.Fatal(err) } } + } // Ensure that integer import with keys runs. diff --git a/holder.go b/holder.go index c87cbe641..f4e61cc48 100644 --- a/holder.go +++ b/holder.go @@ -950,6 +950,10 @@ func (s *holderSyncer) initializeIndexTranslateReplication() error { // Build a map of partition offsets to stream from. m := make(TranslateOffsetMap) for _, index := range s.Holder.Indexes() { + if !index.Keys() { + continue + } + for partitionID := 0; partitionID < s.Cluster.partitionN; partitionID++ { partitionNodes := s.Cluster.partitionNodes(partitionID) isPrimary := partitionNodes[0].ID == node.ID // remote is primary? @@ -967,6 +971,11 @@ func (s *holderSyncer) initializeIndexTranslateReplication() error { } } + // Skip if no replication required. + if len(m) == 0 { + continue + } + // Connect to remote not and begin streaming. rd, err := s.Holder.OpenTranslateReader(context.Background(), node.URI.String(), m) if err != nil { @@ -990,6 +999,9 @@ func (s *holderSyncer) initializeFieldTranslateReplication() error { // Build a map of partition offsets to stream from. m := make(TranslateOffsetMap) for _, index := range s.Holder.Indexes() { + if !index.Keys() { + continue + } for _, field := range index.Fields() { store := field.TranslateStore() offset, err := store.MaxID() @@ -1000,6 +1012,11 @@ func (s *holderSyncer) initializeFieldTranslateReplication() error { } } + // Skip if no replication required. + if len(m) == 0 { + return nil + } + // Connect to coordinator and begin streaming. coordinator := s.Cluster.coordinatorNode() rd, err := s.Holder.OpenTranslateReader(context.Background(), coordinator.URI.String(), m) diff --git a/server/handler_test.go b/server/handler_test.go index 2f0af0345..cb132fe8d 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -991,7 +991,7 @@ func TestHandler_Endpoints(t *testing.T) { if w.Code != gohttp.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) } - target := []uint64{1, 2, 3} + target := []uint64{162529281, 159383553, 160432129} resp := pilosa.TranslateKeysResponse{} err = cmd.API.Serializer.Unmarshal(w.Body.Bytes(), &resp) if err != nil {