private <T extends EventData, U> void handleChange(Event event, String changeType, Class<T> eventDataClass,
            TableIdProvider<T> tableIdProvider,
            RowsProvider<T, U> rowsProvider, BinlogChangeEmitter<U> changeEmitter)
            throws InterruptedException {
        if (skipEvent) {
            // We can skip this because we should already be at least this far ...
            LOGGER.info("Skipping previously processed row event: {}", event);
            return;
        }
        if (ignoreDmlEventByGtidSource) {
            LOGGER.debug("Skipping DML event because this GTID source is filtered: {}", event);
            return;
        }
        final T data = unwrapData(event);
        final TableId tableId = tableIdProvider.getTableId(data);
        final List<U> rows = rowsProvider.getRows(data);

        if (tableId != null && taskContext.getSchema().schemaFor(tableId) != null) {
            int count = 0;
            int numRows = rows.size();
            if (startingRowNumber < numRows) {
                for (int row = startingRowNumber; row != numRows; ++row) {
                    offsetContext.setRowNumber(row, numRows);
                    offsetContext.event(tableId, eventTimestamp);
                    changeEmitter.emit(tableId, rows.get(row));
                    count++;
                }
                if (LOGGER.isDebugEnabled()) {
                    if (startingRowNumber != 0) {
                        LOGGER.debug("Emitted {} {} record(s) for last {} row(s) in event: {}",
                                count, changeType, numRows - startingRowNumber, event);
                    } else {
                        LOGGER.debug("Emitted {} {} record(s) for event: {}", count, changeType, event);
                    }
                }
                offsetContext.changeEventCompleted();
            } else {
                // All rows were previously processed ...
                LOGGER.debug("Skipping previously processed {} event: {}", changeType, event);
            }
        } else {
            informAboutUnknownTableIfRequired(event, tableId, changeType + " row");
        }
        startingRowNumber = 0;
    }
Generate source records for the supplied event with an {@link DeleteRowsEventData}.

@param event the database change data event to be processed; may not be null
@throws InterruptedException if this thread is interrupted while blocking
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java