public static String convertEventToStr(Event event, List<RecordData.FieldGetter> fieldGetters) {
        if (event instanceof SchemaChangeEvent) {
            return event.toString();
        } else if (event instanceof DataChangeEvent) {
            DataChangeEvent dataChangeEvent = (DataChangeEvent) event;
            String eventStr =
                    "DataChangeEvent{"
                            + "tableId="
                            + dataChangeEvent.tableId()
                            + ", before="
                            + getFields(fieldGetters, dataChangeEvent.before())
                            + ", after="
                            + getFields(fieldGetters, dataChangeEvent.after())
                            + ", op="
                            + dataChangeEvent.op()
                            + ", meta="
                            + dataChangeEvent.describeMeta()
                            + '}';
            return eventStr;
        }
        return "Event{}";
    }
@Test
    public void testConvertEventToStr() {
        Schema schema =
                Schema.newBuilder()
                        .physicalColumn("col1", DataTypes.STRING())
                        .physicalColumn("col2", DataTypes.STRING())
                        .primaryKey("col1")
                        .build();
        TableId tableId = TableId.parse("default.default.table1");
        BinaryRecordDataGenerator generator =
                new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(), DataTypes.STRING()));

        List<RecordData.FieldGetter> fieldGetters = SchemaUtils.createFieldGetters(schema);
        Assert.assertEquals(
                "CreateTableEvent{tableId=default.default.table1, schema=columns={`col1` STRING,`col2` STRING}, primaryKeys=col1, options=()}",
                ValuesDataSinkHelper.convertEventToStr(
                        new CreateTableEvent(tableId, schema), fieldGetters));

        DataChangeEvent insertEvent =
                DataChangeEvent.insertEvent(
                        tableId,
                        generator.generate(
                                new Object[] {
                                    BinaryStringData.fromString("1"),
                                    BinaryStringData.fromString("1")
                                }));
        Assert.assertEquals(
                "DataChangeEvent{tableId=default.default.table1, before=[], after=[1, 1], op=INSERT, meta=()}",
                ValuesDataSinkHelper.convertEventToStr(insertEvent, fieldGetters));
        DataChangeEvent deleteEvent =
                DataChangeEvent.deleteEvent(
                        tableId,
                        generator.generate(
                                new Object[] {
                                    BinaryStringData.fromString("1"),
                                    BinaryStringData.fromString("1")
                                }));
        Assert.assertEquals(
                "DataChangeEvent{tableId=default.default.table1, before=[1, 1], after=[], op=DELETE, meta=()}",
                ValuesDataSinkHelper.convertEventToStr(deleteEvent, fieldGetters));
        DataChangeEvent updateEvent =
                DataChangeEvent.updateEvent(
                        tableId,
                        generator.generate(
                                new Object[] {
                                    BinaryStringData.fromString("1"),
                                    BinaryStringData.fromString("1")
                                }),
                        generator.generate(
                                new Object[] {
                                    BinaryStringData.fromString("1"),
                                    BinaryStringData.fromString("x")
                                }));
        Assert.assertEquals(
                "DataChangeEvent{tableId=default.default.table1, before=[1, 1], after=[1, x], op=UPDATE, meta=()}",
                ValuesDataSinkHelper.convertEventToStr(updateEvent, fieldGetters));
    }