From 321fa24831409ffbd829d3fafaba64102196f9c4 Mon Sep 17 00:00:00 2001
From: HHans09 <heena.hans@molecula.com>
Date: Thu, 22 Sep 2022 11:11:46 -0400
Subject: [PATCH] FB-1646 : UPdated code post code review

---
 idk/kafka/source.go | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/idk/kafka/source.go b/idk/kafka/source.go
index 55fd4158e..f85dd043e 100644
--- a/idk/kafka/source.go
+++ b/idk/kafka/source.go
@@ -232,6 +232,11 @@ func (r *Record) Commit(ctx context.Context) error {
 		p = x.Partition
 	}
 
+	_, err := r.src.CommitMessages(r.src.highmarks)
+	if err != nil {
+		return errors.Wrap(err, "failed to commit messages")
+	}
+
 	r.src.spool = remaining
 	r.src.spoolBase = idx
 	return nil
@@ -292,7 +297,8 @@ func (s *Source) Open() error {
 			return err
 		}
 		s.Log.Debugf(buf.String())
-
+		stv, iv := confluent.LibraryVersion()
+		s.Log.Debugf("version:(%v) %v", iv, stv)
 	}
 	cl, err := confluent.NewConsumer(s.ConfigMap)
 	if err != nil {