From 08888a1a6848c5fdf2d9288fd6dc3dcd36f189da Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Toni=20Ram=C3=ADrez?=
 <58293609+ToniRamirezM@users.noreply.github.com>
Date: Thu, 17 Oct 2024 13:36:01 +0200
Subject: [PATCH] feat: currentStreamBatchRaw sanity check (#131) (#132)

---
 aggregator/aggregator.go | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/aggregator/aggregator.go b/aggregator/aggregator.go
index 2003f0e2..7106b615 100644
--- a/aggregator/aggregator.go
+++ b/aggregator/aggregator.go
@@ -425,6 +425,12 @@ func (a *Aggregator) handleReceivedDataStream(
 
 			switch entry.Type {
 			case datastreamer.EntryType(datastream.EntryType_ENTRY_TYPE_BATCH_START):
+				// Check currentStreamBatchRaw is empty as sanity check
+				if len(a.currentStreamBatchRaw.Blocks) > 0 {
+					a.logger.Errorf("currentStreamBatchRaw should be empty, "+
+						"but it contains %v blocks", len(a.currentStreamBatchRaw.Blocks))
+					a.resetCurrentBatchData()
+				}
 				batch := &datastream.BatchStart{}
 				err := proto.Unmarshal(entry.Data, batch)
 				if err != nil {