import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData; //导入依赖的package包/类
private List<Record> toRecords(Table table,
                               EventHeader eventHeader,
                               DeleteRowsEventData eventData,
                               SourceOffset offset) {
  List<Record> res = new ArrayList<>(eventData.getRows().size());
  for (Serializable[] row : eventData.getRows()) {
    Record record = recordFactory.create(offset.format());
    Map<String, Field> fields = createHeader(table, eventHeader, offset);
    fields.put(TYPE_FIELD, create("DELETE"));
    record.getHeader().setAttribute(
        OperationType.SDC_OPERATION_TYPE,
        String.valueOf(OperationType.DELETE_CODE)
    );
    List<ColumnValue> columnValues = zipColumnsValues(eventData.getIncludedColumns(), table, row);
    Map<String, Field> data = toMap(columnValues);
    fields.put(OLD_DATA_FIELD, create(data));

    record.set(create(fields));
    res.add(record);
  }
  return res;
}
 

import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData; //导入依赖的package包/类
void dispatch(Event evt) {
  Object data = evt.getData();
  if (TableMapEventData.class.isInstance(data)) {
    handleTableMapEvent((TableMapEventData) data);
  } else if (EventType.isWrite(evt.getHeader().getEventType())) {
    handleWriteEvent((WriteRowsEventData) data);
  } else if (EventType.isUpdate(evt.getHeader().getEventType())) {
    handleUpdateEvent((UpdateRowsEventData) data);
  } else if (EventType.isDelete(evt.getHeader().getEventType())) {
    handleDeleteEvent((DeleteRowsEventData) data);
  } else if (QueryEventData.class.isInstance(data)) {
    handleQueryEvent((QueryEventData) data);
  }
}
 

import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData; //导入依赖的package包/类
private void handleDeleteEvent(DeleteRowsEventData evt) {
  if (lastTableMap == null) {
    throw new IllegalStateException("Missing table map event");
  }
  evt.getRows()
     .forEach(row ->
       handleRowEvent(lastTableMap.getDatabase(), lastTableMap.getTable(),
         "delete", Arrays.asList(row))
     );
  lastTableMap = null;
}
 

import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData; //导入依赖的package包/类
private void handleDeleteRows(DeleteRowsEventData data, List<DataChange> dataChanges) throws JsonProcessingException {
    System.out.println(data);
    TableMapEventData tableData = tableIdToDataMap.get(data.getTableId());
    Pair<String, String> tableUniqueName = new ImmutablePair<String, String>(tableData.getDatabase(), 
            tableData.getTable());
    
    TableMetadata tableMetadata = databaseNameToTableMetadata.get(tableUniqueName);
    // XXX Error handling if tableMetadata == null
    ArrayList<String> columnNames = tableMetadata.getColumnNames();
    List<Integer> primaryKeyIndices = tableMetadata.getPrimaryKeyIndices();
    
    List<Serializable[]> rows = data.getRows();
    for (Serializable[] row: rows) {
        
        Map<String, Serializable> primaryKeyMap = new HashMap<String, Serializable>();
        for (int index : primaryKeyIndices) {
            String columnName = columnNames.get(index);
            Serializable value = row[index];
            primaryKeyMap.put(columnName, value);
        }
       
        DataChange change = new DataChange();
        change.setPayload(null);
        change.setPrimaryKey(primaryKeyMap);
        change.setTableName(tableUniqueName);
        
        dataChanges.add(change);
    }
}
 

import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData; //导入依赖的package包/类
public int getColumnCount() {
    if (this.binlogEventV4 != null) {
        if (binlogEventV4.getHeader().getEventType() == MySQLConstants.DELETE_ROWS_EVENT) {
            return ((DeleteRowsEvent) binlogEventV4).getColumnCount().intValue();
        }
        else {
            return ((DeleteRowsEventV2) binlogEventV4).getColumnCount().intValue();
        }
    }
    else {
        BitSet includedColumns = ((DeleteRowsEventData) binlogConnectorEvent.getData()).getIncludedColumns();
        return includedColumns.cardinality();
    }
}
 

import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData; //导入依赖的package包/类
public long getTableId() {
    if (this.binlogEventV4 != null) {
        if (binlogEventV4.getHeader().getEventType() == MySQLConstants.DELETE_ROWS_EVENT) {
            return ((DeleteRowsEvent) binlogEventV4).getTableId();
        }
        else {
            return ((DeleteRowsEventV2) binlogEventV4).getTableId();
        }
    }
    else {
        return ((DeleteRowsEventData) binlogConnectorEvent.getData()).getTableId();
    }
}
 

import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData; //导入依赖的package包/类
@Override
public void onEvent(Event event) {
    try {

        EventData data = event.getData();

        if (! (event.getHeader() instanceof EventHeaderV4)) {
            throw new RuntimeException("Header is not V4!");
        }
        System.out.println(event);  

        EventHeaderV4 header = event.getHeader();

        List<DataChange> dataChanges = new ArrayList<DataChange>();
        
        EventType type = header.getEventType();
        if (type == EventType.TABLE_MAP) {
            handleTableMap((TableMapEventData) data);
        } else if (EventType.isWrite(type)) {
            handleWriteRows((WriteRowsEventData) data, dataChanges);
        } else if (EventType.isDelete(type)) {
            handleDeleteRows((DeleteRowsEventData) data, dataChanges);
        } else if (EventType.isUpdate(type)) {
            handleUpdateRows((UpdateRowsEventData) data, dataChanges);
        } else if (type == EventType.QUERY) {
            handleQuery((QueryEventData) data, header);
        } else if (type == EventType.ROTATE) {
            handleRotate((RotateEventData) data);
        } else if (type == EventType.XID) {
            handleXid((XidEventData) data, header);
        }

        for (DataChange change : dataChanges) {
            change.setOffset(this.transactionStartPosition,
                    this.binlogFilename, header.getPosition());
            change.setServerId(this.mysqlServerId);
            outputter.output(change);
        }
        
        outputter.flush();
        
    } catch (Throwable e) {
        // during development, on any and all errors, disconnect.
        // This allows me to learn what errors will happen, and how to
        // handle each one appropriately.
        e.printStackTrace();
        try {
            this.binlogClient.disconnect();
        } catch (IOException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        }

    }
}
 

import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData; //导入依赖的package包/类
public void testOnEventDeleteRows() {
	BinlogEventListener binlogEventListener = new BinlogEventListener(createBinaryLogClient(), createKafkaProducer());
	
	binlogEventListener.onEvent(createTableMapEvent(1, "database", "table"));
	
	EventHeaderV4 eventHeader = new EventHeaderV4();
	eventHeader.setEventType(EventType.EXT_DELETE_ROWS);
	
	DeleteRowsEventData eventData = new DeleteRowsEventData();
	eventData.setTableId(1);
	
	List<Serializable[]> rows = new ArrayList<Serializable[]>();
	
	Serializable[] row = { "1" };
	
	rows.add(row);
	
	eventData.setRows(rows);
	
	binlogEventListener.onEvent(new Event(eventHeader, eventData));
}
 

import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData; //导入依赖的package包/类
@Override
public void onEvent(Event event) {
  LOG.trace("Received event {}", event);
  EventType eventType = event.getHeader().getEventType();
  currentBinLogFileName = client.getBinlogFilename();
  switch (eventType) {
    case TABLE_MAP:
      handleTableMappingEvent((TableMapEventData) event.getData());
      break;
    case PRE_GA_WRITE_ROWS:
    case WRITE_ROWS:
    case EXT_WRITE_ROWS:
      handleRowEvent(event, event.<WriteRowsEventData>getData().getTableId());
      break;
    case PRE_GA_UPDATE_ROWS:
    case UPDATE_ROWS:
    case EXT_UPDATE_ROWS:
      handleRowEvent(event, event.<UpdateRowsEventData>getData().getTableId());
      break;
    case PRE_GA_DELETE_ROWS:
    case DELETE_ROWS:
    case EXT_DELETE_ROWS:
      handleRowEvent(event, event.<DeleteRowsEventData>getData().getTableId());
      break;
    case QUERY:
      QueryEventData queryEventData = event.getData();
      String query = queryEventData.getSql();
      if (isCommit(query)) {
        finishTx();
      } else if (isSchemaChangeQuery(query)) {
        schemaRepository.evictAll();
      }
      break;
    case XID:
      finishTx();
      break;
    case GTID:
      GtidEventData eventData = event.getData();
      currentGtidSet = client.getGtidSet();
      currentTxGtid = eventData.getGtid();
      currentTxEventSeqNo = 0;
      LOG.trace("Started new tx, gtid: {}", currentTxGtid);
      break;
    default:
      // ignore
      break;
  }
}