JavaRDD<Tuple2<List<Row>,SourceParam>> scanRDDs = context.parallelize(Collections.emptyList());
for(DefinitionSource eachSource : definition.getSources()){
	if(!source.isExcludeDelta()){
			scanRDDs.union(context.parallelize(Collections.singletonList(new Tuple2<>(
			lookupService.scan(context,transUtils.tableName(eachSource.getSchemaName(),
			eachSource.getTableName()),param.getTimeSlot(),eachSource.getTimelineKey()).collect(),
			new SourceParam(eachSource,param)))));
	}
}

// Instead of

// List<Tuple2<List<Row>, SourceParam>> scanRDDs = new ArrayList<>();
// 					Definition definition = definitions.getDefinition(param.getName());
// 					scanRDDs.addAll(definition.getSources().parallelStream()
// 					.filter(source -> !source.isExcludeDelta())
// 					.map(source -> new Tuple2<>(lookupService.scan(context,
// 					    transUtils.tableName(source.getSchemaName(),
// 							source.getTableName()), param.getTimeSlot(),
// 							source.getTimelineKey()).collect(), new SourceParam(source, param)))
// 					 .filter(source -> !source._1.isEmpty())
// 					 .collect(Collectors.toList()));

// JavaPairRDD<String, CSVRow> finalRdds = context
// 			.parallelize(scanRDDs, scanRDDs.size())... and so on