--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/stress-tester/src/main/java/com/passus/st/client/netflow/filter/NetflowFilter.java	Wed Mar 25 13:51:03 2020 +0100
@@ -0,0 +1,34 @@
+package com.passus.st.client.netflow.filter;
+
+import com.passus.net.netflow.Netflow;
+import com.passus.st.client.FlowContext;
+import com.passus.st.client.filter.FlowFilter;
+
+public abstract class NetflowFilter implements FlowFilter {
+
+    @Override
+    public int filterInbound(Object req, Object resp, FlowContext context) {
+        if (req instanceof Netflow) {
+            return filterInbound((Netflow) req, resp, context);
+        }
+
+        return DUNNO;
+    }
+
+    public int filterInbound(Netflow req, Object resp, FlowContext context) {
+        return DUNNO;
+    }
+
+    @Override
+    public int filterOutbound(Object req, Object resp, FlowContext context) {
+        if (req instanceof Netflow) {
+            return filterOutbound((Netflow) req, resp, context);
+        }
+
+        return DUNNO;
+    }
+
+    public int filterOutbound(Netflow req, Object resp, FlowContext context) {
+        return DUNNO;
+    }
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/stress-tester/src/main/java/com/passus/st/client/netflow/filter/NetflowVersionFilter.java	Wed Mar 25 13:51:03 2020 +0100
@@ -0,0 +1,97 @@
+package com.passus.st.client.netflow.filter;
+
+import com.passus.commons.Assert;
+import com.passus.commons.annotations.Plugin;
+import com.passus.config.Configuration;
+import com.passus.config.ConfigurationContext;
+import com.passus.config.annotations.NodeDefinitionCreate;
+import com.passus.config.schema.NodeDefinition;
+import com.passus.config.schema.NodeDefinitionCreator;
+import com.passus.net.netflow.Netflow;
+import com.passus.st.client.FlowContext;
+import com.passus.st.client.filter.FlowFilter;
+import com.passus.st.plugin.PluginConstants;
+
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+
+import static com.passus.config.schema.ConfigurationSchemaBuilder.*;
+
+@NodeDefinitionCreate(NetflowVersionFilter.NetflowVersionFilterNodeDefCreator.class)
+@Plugin(name = NetflowVersionFilter.TYPE, category = PluginConstants.CATEGORY_FLOW_FILTER)
+public class NetflowVersionFilter extends NetflowFilter {
+
+    public static final String TYPE = "netflowVersion";
+
+    private int[] versions = new int[0];
+
+    private boolean acceptOnMatch = true;
+
+    public Set<Integer> getVersions() {
+        Set<Integer> out = new HashSet<>();
+        for (int version : versions) {
+            out.add(version);
+        }
+        return out;
+    }
+
+    public void setVersions(Set<Integer> versions) {
+        Assert.notContainsNull(versions);
+
+        this.versions = new int[versions.size()];
+        Iterator<Integer> it = versions.iterator();
+        int i = 0;
+        while (it.hasNext()) {
+            this.versions[i++] = it.next();
+        }
+    }
+
+    public boolean isAcceptOnMatch() {
+        return acceptOnMatch;
+    }
+
+    public void setAcceptOnMatch(boolean acceptOnMatch) {
+        this.acceptOnMatch = acceptOnMatch;
+    }
+
+    @Override
+    public void configure(Configuration config, ConfigurationContext context) {
+        acceptOnMatch = config.getBoolean("acceptOnMatch", true);
+        List<Integer> versions = (List<Integer>) config.get("versions");
+        setVersions(new HashSet<>(versions));
+    }
+
+    @Override
+    public int filterOutbound(Netflow req, Object resp, FlowContext context) {
+        for (int version : versions) {
+            if (version == req.getVersion()) {
+                return acceptOnMatch ? ACCEPT : DENY;
+            }
+        }
+
+        return DUNNO;
+    }
+
+    @Override
+    public FlowFilter instanceForWorker(int index) {
+        NetflowVersionFilter filter = new NetflowVersionFilter();
+        filter.versions = new int[versions.length];
+        System.arraycopy(versions, 0, filter.versions, 0, versions.length);
+        filter.acceptOnMatch = acceptOnMatch;
+        return filter;
+    }
+
+    public static class NetflowVersionFilterNodeDefCreator implements NodeDefinitionCreator {
+
+        @Override
+        public NodeDefinition create() {
+            return mapDef(
+                    tupleDef("acceptOnMatch", valueDefBool()).setRequired(false),
+                    tupleDef("versions", listDef(valueDefInteger()))
+            );
+        }
+
+    }
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/stress-tester/src/test/java/com/passus/st/client/netflow/filter/NetflowVersionFilterTest.java	Wed Mar 25 13:51:03 2020 +0100
@@ -0,0 +1,42 @@
+package com.passus.st.client.netflow.filter;
+
+import com.passus.config.NodeException;
+import com.passus.config.validation.Errors;
+import com.passus.st.client.filter.FlowFilter;
+import com.passus.st.client.filter.FlowFiltersConfigurator;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+
+import static com.passus.st.utils.Assert.assertNoError;
+import static org.testng.AssertJUnit.assertEquals;
+import static org.testng.AssertJUnit.assertTrue;
+
+public class NetflowVersionFilterTest {
+
+    @Test
+    public void testConfigure() throws IOException, NodeException {
+        String filterConfig = "filters:\n"
+                + "    - type: netflowVersion\n"
+                + "      acceptOnMatch: true\n"
+                + "      versions: [5, 9]\n";
+
+        Errors errors = new Errors();
+        List<FlowFilter> filters = FlowFiltersConfigurator.getFilters(filterConfig, errors, null);
+
+        errors.getAllErrors().forEach(System.out::println);
+        assertNoError(errors);
+        assertEquals(1, filters.size());
+        assertTrue(filters.get(0) instanceof NetflowVersionFilter);
+
+        NetflowVersionFilter filter = (NetflowVersionFilter) filters.get(0);
+        assertTrue(filter.isAcceptOnMatch());
+        Set<Integer> versions = filter.getVersions();
+        assertEquals(2, versions.size());
+        assertTrue(versions.contains(5));
+        assertTrue(versions.contains(9));
+    }
+
+}
\ No newline at end of file