added threshold for running ftd and match

common/core/src/test/java/zingg/common/core/executor/FtdLabelCombinedExecutorTester.java

@@ -0,0 +1,65 @@
+package zingg.common.core.executor;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import zingg.common.client.ClientOptions;
+import zingg.common.client.ZFrame;
+import zingg.common.client.ZinggClientException;
+import zingg.common.core.executor.validate.ExecutorValidator;
+
+import java.io.IOException;
+
+public class FtdLabelCombinedExecutorTester<S, D, R, C, T> extends ExecutorTester<S, D, R, C, T> {
+
+    private static final Log LOG = LogFactory.getLog(FtdLabelCombinedExecutorTester.class);
+    private final ZinggBase<S, D, R, C, T> labelExecutor;
+    private final ExecutorValidator<S, D, R, C, T> labelValidator;
+    //max how many times we want to run
+    //to avoid infinite loops in rare cases
+    private int MAX_RUN_COUNT = 5;
+    private int REQUIRED_MATCH_COUNT = 10;
+
+    //setting labeller properties here
+    //ftd properties are already set by super
+    public FtdLabelCombinedExecutorTester(ZinggBase<S, D, R, C, T> ftdExecutor, ExecutorValidator<S, D, R, C, T> ftdValidator, String configFile,
+                                          ZinggBase<S, D, R, C, T> labelExecutor, ExecutorValidator<S, D, R, C, T> labelValidator) throws ZinggClientException, IOException {
+        super(ftdExecutor, ftdValidator, configFile);
+        this.labelExecutor = labelExecutor;
+        this.labelValidator = labelValidator;
+    }
+
+    //need to execute until we get
+    //required number of matches
+    @Override
+    public void initAndExecute(S session) throws ZinggClientException {
+        executor.init(args,session, new ClientOptions());
+        labelExecutor.init(args, session, new ClientOptions());
+        runUntilThreshold();
+    }
+
+    @Override
+    public void validateResults() throws ZinggClientException {
+        validator.validateResults();
+        labelValidator.validateResults();
+    }
+
+    //run until max run count reached or
+    //required match count reached, whichever
+    //reaches earlier
+    private void runUntilThreshold() throws ZinggClientException {
+        long matchCount = 0;
+        while(MAX_RUN_COUNT > 0 && matchCount < REQUIRED_MATCH_COUNT) {
+            executor.execute();
+            labelExecutor.execute();
+            ZFrame<D, R, C> markedRecords = labelExecutor.getMarkedRecords();
+            matchCount = getMatchRecordCount(markedRecords);
+            MAX_RUN_COUNT--;
+        }
+        LOG.info("total number of matches discovered, " + matchCount);
+    }
+
+    private long getMatchRecordCount(ZFrame<D, R, C> markedRecords) {
+        return markedRecords.filter(markedRecords.equalTo("z_isMatch", 1.0)).count();
+    }
+
+}

common/core/src/test/java/zingg/common/core/executor/TestExecutorsSingle.java

@@ -27,16 +27,9 @@ public TestExecutorsSingle(){
 	@Override
 	public List<ExecutorTester<S, D, R, C, T>> getExecutors() throws ZinggClientException, IOException{
 	    TrainingDataFinder<S, D, R, C, T> tdf = getTrainingDataFinder();
-		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(tdf, new TrainingDataFinderValidator<S, D, R, C, T>(tdf),getConfigFile()));
-
 		Labeller<S, D, R, C, T> labeler = getLabeller();
-		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(labeler, new LabellerValidator<S, D, R, C, T>(labeler),getConfigFile()));
-
-		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(tdf, new TrainingDataFinderValidator<S, D, R, C, T>(tdf),getConfigFile()));
-		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(labeler, new LabellerValidator<S, D, R, C, T>(labeler),getConfigFile()));
-
-		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(tdf, new TrainingDataFinderValidator<S, D, R, C, T>(tdf),getConfigFile()));
-		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(labeler, new LabellerValidator<S, D, R, C, T>(labeler),getConfigFile()));
+		executorTesterList.add(new FtdLabelCombinedExecutorTester<S, D, R, C, T>(tdf, new TrainingDataFinderValidator<S, D, R, C, T>(tdf), getConfigFile(),
+				labeler, new LabellerValidator<S, D, R, C, T>(labeler)));
 
 		Trainer<S, D, R, C, T> trainer = getTrainer();
 		executorTesterList.add(new ExecutorTester<S, D, R, C, T>(trainer,getTrainerValidator(trainer),getConfigFile()));