}
final EventProcessor eventProcessor = extensionFactory.getExtension(EventProcessor.class,
dataMediaPair.getFilterData());
if (eventProcessor instanceof DataSourceFetcherAware) {
((DataSourceFetcherAware) eventProcessor).setDataSourceFetcher(new DataSourceFetcher() {
@Override
public DataSource fetch(Long tableId) {
DataMedia dataMedia = ConfigHelper.findDataMedia(pipeline, tableId);
return dataSourceService.getDataSource(pipeline.getId(), dataMedia.getSource());