Penalize flow file in ExecuteSparkJob if Kylo is offline
@@ -573,7 +573,15 @@ public class ExecuteSparkJob extends BaseProcessor {
for (final String id : dataSetIds.split(",")) {
dataSets.append((dataSets.length() == 0) ? '[' : ',');
- final Optional<com.thinkbiganalytics.kylo.catalog.rest.model.DataSet> dataSet = provider.getDataSet(id);
+ final Optional<com.thinkbiganalytics.kylo.catalog.rest.model.DataSet> dataSet;
+ try {
+ dataSet = provider.getDataSet(id);
+ } catch (final Exception e) {
+ getLog().error("Unable to access data set: {}: {}", new Object[]{id, e}, e);
+ session.penalize(flowFile);
+ session.transfer(flowFile);
+ return null;
+ }
if (dataSet.isPresent()) {
if (dataSet.get().getJars() != null) {
extraJarPaths.addAll(dataSet.get().getJars());
@@ -613,7 +621,15 @@ public class ExecuteSparkJob extends BaseProcessor {
if (!resolvedDatasources.contains(id)) {
datasources.append((datasources.length() == 0) ? '[' : ',');
- final Optional<Datasource> datasource = provider.getDatasource(id);
+ final Optional<Datasource> datasource;
+ try {
+ datasource = provider.getDatasource(id);
+ } catch (final Exception e) {
+ getLog().error("Unable to access data source: {}: {}", new Object[]{id, e}, e);
+ session.penalize(flowFile);
+ session.transfer(flowFile);
+ return null;
+ }
if (datasource.isPresent()) {
if (datasource.get() instanceof JdbcDatasource && StringUtils.isNotBlank(((JdbcDatasource) datasource.get()).getDatabaseDriverLocation())) {
final String[] databaseDriverLocations = ((JdbcDatasource) datasource.get()).getDatabaseDriverLocation().split(",");
