'commit'

src/main/java/UnitTest/OpenGaussReplicationToKafka.java

@@ -174,6 +174,9 @@ public class OpenGaussReplicationToKafka {
                 ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, res);
                 kafkaProducer.send(record);
                 System.out.println("send ok ==> " + res);
+                stream.setAppliedLSN(nowLsn);
+                stream.setFlushedLSN(nowLsn);
+                stream.forceUpdateStatus();
                 //write to redis
                 Redis.use().setex(key, EXPIRE_SECONDS, 1);
             }