@Override
      public Collection<M> apply(M message) {
        return new ArrayList<M>() {
          {
            if (filterFn.apply(message)) {
              this.add(message);
            }
          }
        };
      }
The spec for an operator that filters input messages based on some conditions.

@param <M> type of input message
samza-core/src/main/java/org/apache/samza/operators/spec/FilterOperatorSpec.java