class RangerAuthorizationFilter extends FilterBase {

	@SuppressWarnings("deprecation")
	@Override
	public ReturnCode filterKeyValue(Cell kv) throws IOException {

		if (LOG.isDebugEnabled()) {
			LOG.debug("==> filterKeyValue");
		}

		String family = null;
		byte[] familyBytes = kv.getFamily();
		if (familyBytes != null && familyBytes.length > 0) {
			family = Bytes.toString(familyBytes);
			if (LOG.isDebugEnabled()) {
				LOG.debug("filterKeyValue: evaluating family[" + family + "].");
			}
		}
		String column = null;
		if (kv.getQualifier() != null && kv.getQualifier().length > 0) {
			column = Bytes.toString(kv.getQualifier());
			if (LOG.isDebugEnabled()) {
				LOG.debug("filterKeyValue: evaluating column[" + column + "].");
			}
		} else {
			LOG.warn("filterKeyValue: empty/null column set! Unexpected!");
		}

		ReturnCode result = ReturnCode.NEXT_COL;
		boolean authCheckNeeded = false;
		if (family == null) {
			LOG.warn("filterKeyValue: Unexpected - null/empty family! Access denied!");
		} else if (_familiesAccessDenied.contains(family)) {
			LOG.debug("filterKeyValue: family found in access denied families cache.  Access denied.");
		} else if (_columnsAccessAllowed.containsKey(family)) {
			LOG.debug("filterKeyValue: family found in column level access results cache.");
			if (_columnsAccessAllowed.get(family).contains(column)) {
				LOG.debug("filterKeyValue: family/column found in column level access results cache. Access allowed.");
				result = ReturnCode.INCLUDE;
			} else {
				LOG.debug("filterKeyValue: family/column not in column level access results cache. Access denied.");
			}
		} else if (_familiesAccessAllowed.contains(family)) {
			LOG.debug("filterKeyValue: family found in access allowed families cache.  Must re-authorize for correct audit generation.");
			authCheckNeeded = true;
		} else if (_familiesAccessIndeterminate.contains(family)) {
			LOG.debug("filterKeyValue: family found in indeterminate families cache.  Evaluating access...");
			authCheckNeeded = true;
		} else {
			LOG.warn("filterKeyValue: Unexpected - alien family encountered that wasn't seen by pre-hook!  Access Denied.!");
		}

		if (authCheckNeeded) {
			LOG.debug("filterKeyValue: Checking authorization...");
			_session.columnFamily(family)
					.column(column)
					.buildRequest()
					.authorize();
			// must always purge the captured audit event out of the audit handler to avoid messing up the next check
			AuthzAuditEvent auditEvent = _auditHandler.getAndDiscardMostRecentEvent();
			if (_session.isAuthorized()) {
				LOG.debug("filterKeyValue: Access granted.");
				result = ReturnCode.INCLUDE;
				if (auditEvent != null) {
					LOG.debug("filterKeyValue: access is audited.");
					_auditHandler.logAuthzAudits(Collections.singletonList(auditEvent));
				} else {
					LOG.debug("filterKeyValue: no audit event returned.  Access not audited.");
				}
			} else {
				LOG.debug("filterKeyValue: Access denied.  Denial not audited.");
			}
		}
		if (LOG.isDebugEnabled()) {
			LOG.debug("filterKeyValue: " + result);
		}
		return result;
	}

	public  RangerAuthorizationFilter(AuthorizationSession session, Set<String> familiesAccessAllowed, Set<String> familiesAccessDenied, Set<String> familiesAccessIndeterminate,
									 Map<String, Set<String>> columnsAccessAllowed);

}