@Override
    public boolean apply(SamzaSqlRelMessage message) {
      long startProcessing = System.nanoTime();
      Object[] result = new Object[1];
      try {
        expr.execute(translatorContext.getExecutionContext(), context, translatorContext.getDataContext(),
            message.getSamzaSqlRelRecord().getFieldValues().toArray(), result);
      } catch (Exception e) {
        String errMsg = String.format("Handling the following rel message ran into an error. %s", message);
        LOG.error(errMsg, e);
        throw new SamzaException(errMsg, e);
      }
      if (result[0] == null) {
        // Case filter is applied on a null value -> result is neither true or false.
        // Samza Filter operator supports primitive return types only, return false as per current convention.
        return false;
      }
      if (result[0] instanceof Boolean) {
        boolean retVal = (Boolean) result[0];
        LOG.debug(
            String.format("return value for input %s is %s",
                Arrays.asList(message.getSamzaSqlRelRecord().getFieldValues()).toString(), retVal));
        updateMetrics(startProcessing, retVal, System.nanoTime());
        return retVal;
      } else {
        LOG.error("return value is not boolean for rel message: {}", message);
        return false;
      }
    }
FilterTranslatorFunction to process input events, apply the filter and produce output
events accordingly
samza-sql/src/main/java/org/apache/samza/sql/translator/FilterTranslator.java