Search before asking

• I searched in the issues and found nothing similar.

Fluss version

0.5.0

Minimal reproduce step

Add a test in FlussTableITCase

   @Test
    void testPut() throws Exception {
        createTable(DATA1_TABLE_PATH_PK, DATA1_TABLE_INFO_PK.getTableDescriptor(), false);
        Table table = conn.getTable(DATA1_TABLE_PATH_PK);

        TableDescriptor tableDescriptor = table.getDescriptor();
        UpsertWriter upsertWriter = table.getUpsertWriter();

        InternalRow row =
                compactedRow(tableDescriptor.getSchema().toRowType(), new Object[] {1, "2"});
        upsertWriter.delete(row);
        upsertWriter.flush();

        upsertWriter.upsert(row).get();
    }

What doesn't meet your expectations?

It shouldn't throw excpetion in the second put although the first deletion delete nothing

Anything else?

• In the first put with batch sequence, it delete a non-exist key, no binlog is generated which will cause skipping to append to log, so the currently sequene id is still -1;
• In the second put with batch sequence 1, it'll then append to the log, which cause sequene not continuous. So the exception is thrown

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!