/*
 * Copyright (C) 2014-2015 LinkedIn Corp. All rights reserved.
 *
 * Licensed 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.
 */

package gobblin.policies.count;

import gobblin.configuration.ConfigurationKeys;
import gobblin.configuration.State;
import gobblin.qualitychecker.task.TaskLevelPolicy;


public class RowCountRangePolicy extends TaskLevelPolicy {
  private final long rowsRead;
  private final long rowsWritten;
  private final double range;

  public RowCountRangePolicy(State state, Type type) {
    super(state, type);
    this.rowsRead = state.getPropAsLong(ConfigurationKeys.EXTRACTOR_ROWS_EXPECTED);
    this.rowsWritten = state.getPropAsLong(ConfigurationKeys.WRITER_ROWS_WRITTEN);
    this.range = state.getPropAsDouble(ConfigurationKeys.ROW_COUNT_RANGE);
  }

  @Override
  public Result executePolicy() {
    if (Math.abs((this.rowsWritten - this.rowsRead) / (double)this.rowsRead) <= this.range) {
      return Result.PASSED;
    } else {
      return Result.FAILED;
    }
  }
}

@@ -541,16 +541,74 @@
 count;%0A%0A
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0A
 import g
@@ -892,16 +892,105 @@
 range;%0A%0A
+  private static final Logger LOG = LoggerFactory.getLogger(RowCountRangePolicy.class);%0A%0A
   public
@@ -1361,12 +1361,31 @@
     
-if (
+double computedRange = 
 Math
@@ -1448,16 +1448,39 @@
 owsRead)
+;%0A    if (computedRange
  %3C= this
@@ -1530,16 +1530,255 @@
  else %7B%0A
+%0A      LOG.info(String%0A          .format(%0A              %22RowCountRangePolicy check failed. Rows read %25s, Rows written %25s, computed range %25s, expected range %25s %22,%0A              this.rowsRead, this.rowsWritten, computedRange, this.range));%0A%0A
       re

Add logs when quality check fails