broker/consumer: add CreateRevision to ListResponse

CreateRevision is the revision at which the JournalSpec or ShardSpec was
created within Etcd, and pairs with the ModRevision which is already
surfaced in the API response.

broker/list_apply_api.go

@@ -6,7 +6,7 @@ import (
 	"strings"
 
 	log "github.com/sirupsen/logrus"
-	"go.etcd.io/etcd/client/v3"
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.gazette.dev/core/allocator"
 	pb "go.gazette.dev/core/broker/protocol"
 	pbx "go.gazette.dev/core/broker/protocol/ext"
@@ -61,6 +61,7 @@ func (svc *Service) List(ctx context.Context, req *pb.ListRequest) (resp *pb.Lis
 			continue
 		}
 		journal.ModRevision = s.Items[cur.Left].Raw.ModRevision
+		journal.CreateRevision = s.Items[cur.Left].Raw.CreateRevision
 		pbx.Init(&journal.Route, s.Assignments[cur.RightBegin:cur.RightEnd])
 		pbx.AttachEndpoints(&journal.Route, s.KS)
 

broker/list_apply_api_test.go

@@ -40,6 +40,7 @@ func TestListCases(t *testing.T) {
 	}
 
 	var broker = newTestBroker(t, etcd, pb.ProcessSpec_ID{Zone: "local", Suffix: "broker"})
+	var fixtureRevision int64
 	{
 		var resp, err = broker.client().Apply(ctx, &pb.ApplyRequest{
 			Changes: []pb.ApplyRequest_Change{
@@ -50,6 +51,7 @@ func TestListCases(t *testing.T) {
 		})
 		require.NoError(t, err)
 		require.Equal(t, pb.Status_OK, resp.Status)
+		fixtureRevision = resp.Header.Etcd.Revision
 	}
 
 	// Assign |specC| to |bk|, to verify its returned non-empty Route.
@@ -72,6 +74,9 @@ func TestListCases(t *testing.T) {
 		for i, exp := range expect {
 			require.Equal(t, *exp, resp.Journals[i].Spec)
 
+			require.Equal(t, fixtureRevision, resp.Journals[i].CreateRevision)
+			require.Equal(t, fixtureRevision, resp.Journals[i].ModRevision)
+
 			if exp == specC {
 				require.Equal(t, []pb.ProcessSpec_ID{{Zone: "local", Suffix: "broker"}},
 					resp.Journals[i].Route.Members)

broker/protocol/protocol.proto

@@ -584,10 +584,12 @@ message ListResponse {
   // Journals of the response.
   message Journal {
     JournalSpec spec = 1 [ (gogoproto.nullable) = false ];
-    // Current ModRevision of the JournalSpec.
+    // Current Etcd ModRevision of the JournalSpec.
     int64 mod_revision = 2;
     // Route of the journal, including endpoints.
     Route route = 3 [ (gogoproto.nullable) = false ];
+    // Etcd CreateRevision of the JournalSpec.
+    int64 create_revision = 4;
   }
   repeated Journal journals = 3 [ (gogoproto.nullable) = false ];
 }

consumer/protocol/protocol.proto

@@ -332,12 +332,14 @@ message ListResponse {
   // Shards of the response.
   message Shard {
     ShardSpec spec = 1 [ (gogoproto.nullable) = false ];
-    // Current ModRevision of the ShardSpec.
+    // Current Etcd ModRevision of the ShardSpec.
     int64 mod_revision = 2;
     // Route of the shard, including endpoints.
     protocol.Route route = 3 [ (gogoproto.nullable) = false ];
     // Status of each replica. Cardinality and ordering matches |route|.
     repeated ReplicaStatus status = 4 [ (gogoproto.nullable) = false ];
+    // Etcd CreateRevision of the JournalSpec.
+    int64 create_revision = 5;
   }
   repeated Shard shards = 3 [ (gogoproto.nullable) = false ];
   // Optional extension of the ListResponse.

consumer/shard_api.go

@@ -82,6 +82,7 @@ func ShardList(ctx context.Context, srv *Service, req *pc.ListRequest) (*pc.List
 			continue
 		}
 		shard.ModRevision = s.Items[cur.Left].Raw.ModRevision
+		shard.CreateRevision = s.Items[cur.Left].Raw.CreateRevision
 		pbx.Init(&shard.Route, s.Assignments[cur.RightBegin:cur.RightEnd])
 		pbx.AttachEndpoints(&shard.Route, s.KS)
 

consumer/shard_api_test.go

@@ -69,6 +69,9 @@ func TestAPIListCases(t *testing.T) {
 		for i, exp := range expect {
 			require.Equal(t, *exp, resp.Shards[i].Spec)
 
+			require.NotZero(t, resp.Shards[i].ModRevision)
+			require.NotZero(t, resp.Shards[i].CreateRevision)
+
 			var numAsn int
 			if exp == specB {
 				numAsn = 1