streamingccl: add GenerationEvent type

Add `GenerationEvent` as a possible event type to be emitted over a cluster stream. When a `GenerationEvent` is emitted, we should be able to get its topology as well as the start time of the new generation.

Release note: None

pkg/ccl/streamingccl/event.go

@@ -23,6 +23,9 @@ const (
 	// CheckpointEvent indicates that GetResolved will be meaningful. The resolved
 	// timestamp indicates that all KVs have been emitted up to this timestamp.
 	CheckpointEvent
+	// GenerationEvent indicates that the stream should start ingesting with the
+	// updated topology.
+	GenerationEvent
 )
 
 // Event describes an event emitted by a cluster to cluster stream.  Its Type
@@ -84,6 +87,26 @@ func (ce checkpointEvent) GetResolved() *hlc.Timestamp {
 	return &ce.resolvedTimestamp
 }
 
+// generationEvent indicates that the topology of the stream has changed.
+type generationEvent struct{}
+
+var _ Event = generationEvent{}
+
+// Type implements the Event interface.
+func (ge generationEvent) Type() EventType {
+	return GenerationEvent
+}
+
+// GetKV implements the Event interface.
+func (ge generationEvent) GetKV() *roachpb.KeyValue {
+	return nil
+}
+
+// GetResolved implements the Event interface.
+func (ge generationEvent) GetResolved() *hlc.Timestamp {
+	return nil
+}
+
 // MakeKVEvent creates an Event from a KV.
 func MakeKVEvent(kv roachpb.KeyValue) Event {
 	return kvEvent{kv: kv}
@@ -93,3 +116,8 @@ func MakeKVEvent(kv roachpb.KeyValue) Event {
 func MakeCheckpointEvent(resolvedTimestamp hlc.Timestamp) Event {
 	return checkpointEvent{resolvedTimestamp: resolvedTimestamp}
 }
+
+// MakeGenerationEvent creates an GenerationEvent.
+func MakeGenerationEvent() Event {
+	return generationEvent{}
+}

pkg/ccl/streamingccl/streamclient/client_test.go

@@ -79,6 +79,8 @@ func ExampleClient() {
 				fmt.Printf("%s->%s@%d\n", kv.Key.String(), string(kv.Value.RawBytes), kv.Value.Timestamp.WallTime)
 			case streamingccl.CheckpointEvent:
 				fmt.Printf("resolved %d\n", event.GetResolved().WallTime)
+			case streamingccl.GenerationEvent:
+				fmt.Printf("received generation event")
 			default:
 				panic(fmt.Sprintf("unexpected event type %v", event.Type()))
 			}