Merge pull request #196 from redpanda-data/add-batch-sync-res

Add batch aware sync response method

Author: Jeffail

CHANGELOG.md

@@ -3,6 +3,12 @@ Changelog
 
 All notable changes to this project will be documented in this file.
 
+## 4.46.0 - 2025-03-20
+
+### Added
+
+- Go API: New batch aware API for capturing synchronous responses from downstream components. (@Jeffail)
+
 ## 4.45.1 - 2025-03-10
 
 ### Fixed

public/service/message.go

@@ -690,6 +690,22 @@ func (m *Message) AddSyncResponse() error {
 	return transaction.SetAsResponse(message.Batch{m.part})
 }
 
+// WithSyncResponseStore returns a modified message batch and a response store
+// associated with all messages within it. If any message of the batch is sent
+// through a processing pipeline or output there is the potential for sync
+// response components to add messages to the store, which can be consumed once
+// an acknowledgement is received.
+func (b MessageBatch) WithSyncResponseStore() (MessageBatch, *SyncResponseStore) {
+	resStore := transaction.NewResultStore()
+
+	newB := b.Copy()
+	for i, m := range newB {
+		m.part = transaction.AddResultStoreMsg(b[i].part, resStore)
+	}
+
+	return newB, &SyncResponseStore{s: resStore}
+}
+
 // AddSyncResponse attempts to add this batch of messages, in its exact current
 // condition, to the synchronous response destined for the original source input
 // of this data. Synchronous responses aren't supported by all inputs, and so

public/service/message_test.go

@@ -519,3 +519,52 @@ func TestSyncResponse(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, "hello world c", string(data))
 }
+
+func TestSyncResponseBatched(t *testing.T) {
+	batchA := MessageBatch{
+		NewMessage([]byte("hello world a 1")),
+		NewMessage([]byte("hello world a 2")),
+		NewMessage([]byte("hello world a 3")),
+	}
+
+	batchB, storeB := batchA.WithSyncResponseStore()
+	batchB[0].SetBytes([]byte("hello world b 1"))
+	batchB[1].SetBytes([]byte("hello world b 2"))
+	batchB[2].SetBytes([]byte("hello world b 3"))
+
+	require.Error(t, batchA.AddSyncResponse())
+	require.NoError(t, batchB.AddSyncResponse())
+
+	batchC := batchB.Copy()
+	batchC[1].SetBytes([]byte("hello world c 2"))
+	require.NoError(t, batchC.AddSyncResponse())
+
+	batchD := batchA.Copy()
+	batchD[1].SetBytes([]byte("hello world d 2"))
+	require.Error(t, batchD.AddSyncResponse())
+
+	resBatches := storeB.Read()
+	require.Len(t, resBatches, 2)
+	require.Len(t, resBatches[0], 3)
+	require.Len(t, resBatches[1], 3)
+
+	for i, c := range []string{
+		"hello world b 1",
+		"hello world b 2",
+		"hello world b 3",
+	} {
+		data, err := resBatches[0][i].AsBytes()
+		require.NoError(t, err)
+		assert.Equal(t, c, string(data))
+	}
+
+	for i, c := range []string{
+		"hello world b 1",
+		"hello world c 2",
+		"hello world b 3",
+	} {
+		data, err := resBatches[1][i].AsBytes()
+		require.NoError(t, err)
+		assert.Equal(t, c, string(data))
+	}
+}