/**
 * Copyright 2015 StreamSets Inc.
 *
 * Licensed under the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.streamsets.pipeline.stage.common;

import com.streamsets.pipeline.api.ErrorCode;
import com.streamsets.pipeline.api.OnRecordError;
import com.streamsets.pipeline.api.Record;
import com.streamsets.pipeline.api.Stage;
import com.streamsets.pipeline.api.StageException;
import com.streamsets.pipeline.api.ToErrorContext;
import com.streamsets.pipeline.api.base.OnRecordErrorException;
import com.streamsets.pipeline.api.impl.Utils;

import java.util.List;

import static com.streamsets.pipeline.api.OnRecordError.DISCARD;

public class DefaultErrorRecordHandler implements ErrorRecordHandler {
  private final ToErrorContext toError;
  private final Stage.Context context;

  /**
   * Deprecated constructor to keep backward compatibility.
   *
   * Given context must also implement ToErrorContext (and thus be one of Processor, Source or Target), otherwise
   * a ClassCastException will be thrown.
   */
  @Deprecated
  public DefaultErrorRecordHandler(Stage.Context context) {
    this(context, (ToErrorContext) context);
  }

  /**
   * Proper constructor that separate configuration from error sink.
   *
   * @param context Context of the stage with configuration of what should happen when error record occur.
   * @param toError Error sink into which records will be send if TO_ERROR is configured by user.
   */
  public DefaultErrorRecordHandler(Stage.Context context, ToErrorContext toError) {
    this.context = context;
    this.toError = toError;
  }

  @Override
  public void onError(ErrorCode errorCode, Object... params) throws StageException {
    switch (context.getOnErrorRecord()) {
      case DISCARD:
        break;
      case TO_ERROR:
        context.reportError(errorCode, params);
        break;
      case STOP_PIPELINE:
        throw new StageException(errorCode, params);
      default:
        throw new IllegalStateException(Utils.format("Unknown OnError value '{}'", context.getOnErrorRecord()));
    }
  }

  @Override
  public void onError(OnRecordErrorException error) throws StageException {
    switch (context.getOnErrorRecord()) {
      case DISCARD:
        break;
      case TO_ERROR:
        toError.toError(error.getRecord(), error);
        break;
      case STOP_PIPELINE:
        throw error;
      default:
        throw new IllegalStateException(Utils.format("Unknown OnError value '{}'", context.getOnErrorRecord()), error);
    }
  }

  @Override
  public void onError(List<Record> batch, StageException error) throws StageException {
    switch (context.getOnErrorRecord()) {
      case DISCARD:
        break;
      case TO_ERROR:
        // Add all the records in batch to error since there is no way to figure out which record in batch
        // caused exception.
        for (Record record : batch) {
          toError.toError(record, error);
        }
        break;
      case STOP_PIPELINE:
        throw error;
      default:
        throw new IllegalStateException(Utils.format("Unknown OnError value '{}'", context.getOnErrorRecord()), error);
    }
  }
}

@@ -937,58 +937,8 @@
 de;%0A
-import com.streamsets.pipeline.api.OnRecordError;%0A
 impo
@@ -1235,169 +1235,256 @@
 ls;%0A
-%0A
 import 
-java.util.List;%0A%0Aimport static com.streamsets.pipeline.api.OnRecordError.DISCARD;%0A%0Apublic class DefaultErrorRecordHandler implements ErrorRecordHandler %7B
+org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0Aimport java.util.List;%0A%0Apublic class DefaultErrorRecordHandler implements ErrorRecordHandler %7B%0A%0A  private static final Logger LOG = LoggerFactory.getLogger(DefaultErrorRecordHandler.class);%0A
 %0A  p
@@ -2442,32 +2442,87 @@
 tageException %7B%0A
+    validateGetOnErrorRecord(null, errorCode, params);%0A
     switch (cont
@@ -2964,32 +2964,81 @@
 tageException %7B%0A
+    validateGetOnErrorRecord(error, null, null);%0A
     switch (cont
@@ -3474,24 +3474,73 @@
 Exception %7B%0A
+    validateGetOnErrorRecord(error, null, null);%0A
     switch (
@@ -4058,10 +4058,629 @@
   %7D%0A  %7D%0A
+%0A  private void validateGetOnErrorRecord(Exception ex, ErrorCode errorCode, Object ... params) %7B%0A    if(context.getOnErrorRecord() == null) %7B%0A      if(ex != null) %7B%0A        LOG.error(%22Can't propagate exception to error stream%22, ex);%0A      %7D%0A      if(errorCode != null) %7B%0A        LOG.error(%22Can't propagate error to error stream: %7B%7D with params %7B%7D%22, errorCode, params);%0A      %7D%0A%0A      // SDC-5042 - Add way to Stage.Context to retrieve instance name; using the Context.toString for now%0A      throw new IllegalStateException(Utils.format(%22Component %7B%7D doesn't have configured error record action.%22, context));%0A    %7D%0A  %7D%0A%0A
 %7D%0A

Add NPE protection to DefaultErrorRecordHandler