diff --git a/pkg/apis/proto/source/v1/mockgen.go b/pkg/apis/proto/source/v1/mockgen.go
index 89dad73d..e2ea1193 100644
--- a/pkg/apis/proto/source/v1/mockgen.go
+++ b/pkg/apis/proto/source/v1/mockgen.go
@@ -1,3 +1,3 @@
 package v1
 
-//go:generate mockgen -destination sourcemock/sourcemock.go -package sourcemock github.com/numaproj/numaflow-go/pkg/apis/proto/source/v1 SourceClient,Source_ReadFnClient
+//go:generate mockgen -destination sourcemock/sourcemock.go -package sourcemock github.com/numaproj/numaflow-go/pkg/apis/proto/source/v1 SourceClient,Source_ReadFnClient,Source_AckFnClient
diff --git a/pkg/apis/proto/source/v1/sourcemock/sourcemock.go b/pkg/apis/proto/source/v1/sourcemock/sourcemock.go
index ef7e51a4..cc3674f9 100644
--- a/pkg/apis/proto/source/v1/sourcemock/sourcemock.go
+++ b/pkg/apis/proto/source/v1/sourcemock/sourcemock.go
@@ -1,5 +1,5 @@
 // Code generated by MockGen. DO NOT EDIT.
-// Source: github.com/numaproj/numaflow-go/pkg/apis/proto/source/v1 (interfaces: SourceClient,Source_ReadFnClient)
+// Source: github.com/numaproj/numaflow-go/pkg/apis/proto/source/v1 (interfaces: SourceClient,Source_ReadFnClient,Source_AckFnClient)
 
 // Package sourcemock is a generated GoMock package.
 package sourcemock
@@ -274,3 +274,140 @@ func (mr *MockSource_ReadFnClientMockRecorder) Trailer() *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Trailer", reflect.TypeOf((*MockSource_ReadFnClient)(nil).Trailer))
 }
+
+// MockSource_AckFnClient is a mock of Source_AckFnClient interface.
+type MockSource_AckFnClient struct {
+	ctrl     *gomock.Controller
+	recorder *MockSource_AckFnClientMockRecorder
+}
+
+// MockSource_AckFnClientMockRecorder is the mock recorder for MockSource_AckFnClient.
+type MockSource_AckFnClientMockRecorder struct {
+	mock *MockSource_AckFnClient
+}
+
+// NewMockSource_AckFnClient creates a new mock instance.
+func NewMockSource_AckFnClient(ctrl *gomock.Controller) *MockSource_AckFnClient {
+	mock := &MockSource_AckFnClient{ctrl: ctrl}
+	mock.recorder = &MockSource_AckFnClientMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockSource_AckFnClient) EXPECT() *MockSource_AckFnClientMockRecorder {
+	return m.recorder
+}
+
+// CloseAndRecv mocks base method.
+func (m *MockSource_AckFnClient) CloseAndRecv() (*v1.AckResponse, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CloseAndRecv")
+	ret0, _ := ret[0].(*v1.AckResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CloseAndRecv indicates an expected call of CloseAndRecv.
+func (mr *MockSource_AckFnClientMockRecorder) CloseAndRecv() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloseAndRecv", reflect.TypeOf((*MockSource_AckFnClient)(nil).CloseAndRecv))
+}
+
+// CloseSend mocks base method.
+func (m *MockSource_AckFnClient) CloseSend() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CloseSend")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// CloseSend indicates an expected call of CloseSend.
+func (mr *MockSource_AckFnClientMockRecorder) CloseSend() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloseSend", reflect.TypeOf((*MockSource_AckFnClient)(nil).CloseSend))
+}
+
+// Context mocks base method.
+func (m *MockSource_AckFnClient) Context() context.Context {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Context")
+	ret0, _ := ret[0].(context.Context)
+	return ret0
+}
+
+// Context indicates an expected call of Context.
+func (mr *MockSource_AckFnClientMockRecorder) Context() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Context", reflect.TypeOf((*MockSource_AckFnClient)(nil).Context))
+}
+
+// Header mocks base method.
+func (m *MockSource_AckFnClient) Header() (metadata.MD, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Header")
+	ret0, _ := ret[0].(metadata.MD)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Header indicates an expected call of Header.
+func (mr *MockSource_AckFnClientMockRecorder) Header() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Header", reflect.TypeOf((*MockSource_AckFnClient)(nil).Header))
+}
+
+// RecvMsg mocks base method.
+func (m *MockSource_AckFnClient) RecvMsg(arg0 interface{}) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RecvMsg", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// RecvMsg indicates an expected call of RecvMsg.
+func (mr *MockSource_AckFnClientMockRecorder) RecvMsg(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecvMsg", reflect.TypeOf((*MockSource_AckFnClient)(nil).RecvMsg), arg0)
+}
+
+// Send mocks base method.
+func (m *MockSource_AckFnClient) Send(arg0 *v1.AckRequest) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Send", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Send indicates an expected call of Send.
+func (mr *MockSource_AckFnClientMockRecorder) Send(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Send", reflect.TypeOf((*MockSource_AckFnClient)(nil).Send), arg0)
+}
+
+// SendMsg mocks base method.
+func (m *MockSource_AckFnClient) SendMsg(arg0 interface{}) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SendMsg", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SendMsg indicates an expected call of SendMsg.
+func (mr *MockSource_AckFnClientMockRecorder) SendMsg(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMsg", reflect.TypeOf((*MockSource_AckFnClient)(nil).SendMsg), arg0)
+}
+
+// Trailer mocks base method.
+func (m *MockSource_AckFnClient) Trailer() metadata.MD {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Trailer")
+	ret0, _ := ret[0].(metadata.MD)
+	return ret0
+}
+
+// Trailer indicates an expected call of Trailer.
+func (mr *MockSource_AckFnClientMockRecorder) Trailer() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Trailer", reflect.TypeOf((*MockSource_AckFnClient)(nil).Trailer))
+}