fix replication errors & test races

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.

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)

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 {