PostgresStreamingChangeEventSource's replicationStream flushLsn after closed

Below is my code to reproduce this issue. Between the first 'engine.close();' and the last 'engine.close();', insert some records( I'm using jmeter, insert about 300 records) to the database and the engine will stop with the error on console.   ~public void run() throws IOException, InterruptedException {~ ~final Properties props = new Properties();~ ~props.setProperty("name", "mizzle");~ ~props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");~ ~props.setProperty("offset.storage.file.filename", "/tmp/offsets1.dat");~ ~/* begin connector properties */~ ~props.setProperty("database.hostname", "");~ ~props.setProperty("database.port", "");~ ~props.setProperty("database.user", "");~ ~props.setProperty("database.password", "");~ ~props.setProperty("database.dbname", "");~ ~props.setProperty("offset.flush.interval.ms", "3000");~ ~props.setProperty("slot.name", "debezium_test_c");~ ~props.setProperty("publication.name", "dbz_publication_test_c");~ ~props.setProperty("plugin.name", "pgoutput");~ ~props.setProperty("connector.class", "io.debezium.connector.postgresql.PostgresConnector");~ ~props.setProperty("database.server.name", "my-app-connector");~ ~// Create the engine with this configuration ...~ ~System.out.println(props.toString());~ ~engine = DebeziumEngine.create(Json.class).using(props)~ ~.using(new DebeziumEngine.ConnectorCallback() {~ ~@Override~ ~public void connectorStarted() {~ ~System.out.println("started");~ ~}~ ~@Override~ ~public void connectorStopped() {~ ~try {~ ~System.out.println("stoped");~ ~engine.close();~ ~} catch (IOException e) {~ ~e.printStackTrace();~ ~}~ ~}~ ~}).using((success, message, error) -> {~ ~System.out.println(message);~ ~System.out.println(error);~ ~})~ ~.notifying((records, committer) -> {~ ~try{~ ~for(ChangeEvent<String, String> record : records){~ ~System.out.println(record);~ ~committer.markProcessed(record);~ ~}~ ~committer.markBatchFinished();~ ~}catch (Exception e){~ ~}~ ~})~ ~.build();~ ~// Run the engine asynchronously ...~ ~ExecutorService executor = Executors.newSingleThreadExecutor();~ ~executor.execute(engine);~ ~Thread.sleep(5000);~ ~engine.close();~ ~Thread.sleep(10000);~ ~executor.execute(engine);~ ~Thread.sleep(5000);~ ~engine.close();~