SAMOA-48: Fix double processing of last even in FilterProcessor
diff --git a/samoa-api/src/main/java/org/apache/samoa/learners/InstanceContent.java b/samoa-api/src/main/java/org/apache/samoa/learners/InstanceContent.java
index 7a49985..c4e6b84 100644
--- a/samoa-api/src/main/java/org/apache/samoa/learners/InstanceContent.java
+++ b/samoa-api/src/main/java/org/apache/samoa/learners/InstanceContent.java
@@ -25,6 +25,7 @@
*/
import net.jcip.annotations.Immutable;
+
import org.apache.samoa.core.SerializableInstance;
import org.apache.samoa.instances.Instance;
@@ -61,7 +62,7 @@
* the is training
*/
public InstanceContent(long index, Instance instance,
- boolean isTraining, boolean isTesting) {
+ boolean isTraining, boolean isTesting) {
if (instance != null) {
this.instance = new SerializableInstance(instance);
}
@@ -192,6 +193,11 @@
this.isLast = isLast;
}
-
-
+ @Override
+ public String toString() {
+ return String
+ .format(
+ "InstanceContent [instanceIndex=%s, classifierIndex=%s, evaluationIndex=%s, instance=%s, isTraining=%s, isTesting=%s, isLast=%s]",
+ instanceIndex, classifierIndex, evaluationIndex, instance, isTraining, isTesting, isLast);
+ }
}
diff --git a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/PredictionCombinerProcessor.java b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/PredictionCombinerProcessor.java
index 8e10094..92a209b 100644
--- a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/PredictionCombinerProcessor.java
+++ b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/ensemble/PredictionCombinerProcessor.java
@@ -105,7 +105,6 @@
int instanceIndex = (int) inEvent.getInstanceIndex();
addStatisticsForInstanceReceived(instanceIndex, inEvent.getClassifierIndex(), prediction, 1);
-
if (hasAllVotesArrivedInstance(instanceIndex)) {
DoubleVector combinedVote = this.mapVotesforInstanceReceived.get(instanceIndex);
if (combinedVote == null) {
diff --git a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/FilterProcessor.java b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/FilterProcessor.java
index a55b813..2c81fd0 100644
--- a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/FilterProcessor.java
+++ b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/FilterProcessor.java
@@ -78,7 +78,7 @@
this.waitingInstances++;
if (this.waitingInstances == this.batchSize || instanceContentEvent.isLastEvent()) {
// Send Instances
- InstancesContentEvent outputEvent = new InstancesContentEvent(instanceContentEvent);
+ InstancesContentEvent outputEvent = new InstancesContentEvent();
while (!this.contentEventList.isEmpty()) {
InstanceContentEvent ice = this.contentEventList.remove(0);
outputEvent.add(ice.getInstanceContent());
diff --git a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/ModelAggregatorProcessor.java b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/ModelAggregatorProcessor.java
index ce2d4c4..cad40a7 100644
--- a/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/ModelAggregatorProcessor.java
+++ b/samoa-api/src/main/java/org/apache/samoa/learners/classifiers/trees/ModelAggregatorProcessor.java
@@ -20,6 +20,8 @@
* #L%
*/
+import static org.apache.samoa.moa.core.Utils.maxIndex;
+
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
@@ -39,7 +41,6 @@
import org.apache.samoa.instances.Instances;
import org.apache.samoa.instances.InstancesHeader;
import org.apache.samoa.learners.InstanceContent;
-import org.apache.samoa.learners.InstanceContentEvent;
import org.apache.samoa.learners.InstancesContentEvent;
import org.apache.samoa.learners.ResultContentEvent;
import org.apache.samoa.moa.classifiers.core.AttributeSplitSuggestion;
@@ -50,8 +51,6 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.samoa.moa.core.Utils.maxIndex;
-
/**
* Model Aggegator Processor consists of the decision tree model. It connects to local-statistic PI via attribute stream
* and control stream. Model-aggregator PI sends the split instances via attribute stream and it sends control messages
