perf(go): start realtime projector at latest offset

go/vehicle-gateway/cmd/realtime-api/main.go

@@ -167,13 +167,7 @@ func consumeKafka(ctx context.Context, logger interface {
 	Warn(string, ...any)
 }, registry *metrics.Registry, updater realtimeUpdater, brokers []string) {
 	kafkaTopics := kafkaTopicsFromEnv()
-	reader := kafka.NewReader(kafka.ReaderConfig{
-		Brokers:     brokers,
-		GroupID:     env("KAFKA_GROUP", "go-realtime-api"),
-		GroupTopics: kafkaTopics,
-		MinBytes:    1,
-		MaxBytes:    10e6,
-	})
+	reader := kafka.NewReader(realtimeReaderConfig(brokers, kafkaTopics))
 	defer reader.Close()
 	logger.Info("realtime kafka consumer started", "topics", strings.Join(kafkaTopics, ","))
 	for {
@@ -189,6 +183,17 @@ func consumeKafka(ctx context.Context, logger interface {
 	}
 }
 
+func realtimeReaderConfig(brokers []string, kafkaTopics []string) kafka.ReaderConfig {
+	return kafka.ReaderConfig{
+		Brokers:     brokers,
+		GroupID:     env("KAFKA_GROUP", "go-realtime-api"),
+		GroupTopics: kafkaTopics,
+		MinBytes:    1,
+		MaxBytes:    10e6,
+		StartOffset: kafka.LastOffset,
+	}
+}
+
 func kafkaTopicsFromEnv() []string {
 	defaultTopics := strings.Join([]string{topics.RawGB32960, topics.RawJT808, topics.RawYutongMQTT}, ",")
 	return splitCSV(env("KAFKA_TOPICS", defaultTopics))

go/vehicle-gateway/cmd/realtime-api/main_test.go

@@ -104,6 +104,17 @@ func TestKafkaTopicsFromEnvDefaultsToGoRawTopics(t *testing.T) {
 	}
 }
 
+func TestRealtimeReaderConfigStartsAtLatestOffset(t *testing.T) {
+	cfg := realtimeReaderConfig([]string{"127.0.0.1:9092"}, []string{"vehicle.raw.go.jt808.v1"})
+
+	if cfg.StartOffset != kafka.LastOffset {
+		t.Fatalf("StartOffset = %d, want kafka.LastOffset %d", cfg.StartOffset, kafka.LastOffset)
+	}
+	if got := strings.Join(cfg.GroupTopics, ","); got != "vehicle.raw.go.jt808.v1" {
+		t.Fatalf("GroupTopics = %q", got)
+	}
+}
+
 func TestRealtimeAPIDoesNotExposeDuplicateMileagePointRoute(t *testing.T) {
 	source, err := os.ReadFile("main.go")
 	if err != nil {