Minor bugs fix for FastenServer
@@ -210,8 +210,8 @@ public class FastenServer implements Runnable {
k.getClass().getCanonicalName());
return new FastenKafkaPlugin(consumerProperties, producerProperties, k, skipOffsets,
- outputDirs.get(k.getClass().getSimpleName()),
- outputLinks.get(k.getClass().getSimpleName()));
+ (outputDirs != null) ? outputDirs.get(k.getClass().getSimpleName()) : null,
+ (outputLinks != null) ? outputLinks.get(k.getClass().getSimpleName()) : null);
}).collect(Collectors.toList());
}
@@ -267,14 +267,14 @@ public class FastenServer implements Runnable {
* @param dataWriterPlugins list of Data Writer plugins
*/
private void setBaseDirectory(List<DataWriter> dataWriterPlugins) {
- if (ObjectUtils.allNotNull(baseDir)) {
dataWriterPlugins.forEach((p) -> {
+ if (ObjectUtils.allNotNull(baseDir)) {
p.setBaseDir(baseDir);
- });
} else {
logger.error("Couldn't set a base directory. Make sure that you have "
+ "provided a valid path to base directory.");
}
+ });
}
/**
