?// 清空records集合

? ? records.clear();

? ? // 構(gòu)建消息，表示數(shù)據(jù)已寫入

? ? final WriteMetadataEvent event = WriteMetadataEvent.builder()

? ? ? ? .taskID(taskID)

? ? ? ? .instantTime(instant) // the write instant may shift but the event still use the currentInstant.

? ? ? ? .writeStatus(writeStatus)

? ? ? ? .lastBatch(false)

? ? ? ? .endInput(false)

? ? ? ? .build();

? ? // 發(fā)送消息給coordinator

? ? this.eventGateway.sendEventToCoordinator(event);

? ? writeStatuses.addAll(writeStatus);

? ? return true;