changes
diff --git a/contrib/src/main/java/com/datatorrent/contrib/accumulo/AccumuloOutputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/accumulo/AccumuloOutputOperator.java
new file mode 100644
index 0000000..edf78b4
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/accumulo/AccumuloOutputOperator.java
@@ -0,0 +1,16 @@
+
+package com.datatorrent.contrib.accumulo;
+
+import org.apache.accumulo.core.data.Mutation;
+
+
+public class AccumuloOutputOperator extends AbstractAccumuloOutputOperator<Object>
+{
+
+ @Override
+ public Mutation operationMutation(Object t)
+ {
+ return null;
+ }
+
+}
diff --git a/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraTransactionableOutputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraTransactionableOutputOperator.java
index 4c4e3c5..9bb5d38 100644
--- a/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraTransactionableOutputOperator.java
+++ b/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraTransactionableOutputOperator.java
@@ -64,6 +64,7 @@
@Override
public void processBatch(Collection<T> tuples)
{
+ System.out.println("in process Batch");
BatchStatement batchCommand = store.getBatchCommand();
for(T tuple: tuples)
{
diff --git a/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraOutputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraOutputOperator.java
index b9a3f69..05162a1 100644
--- a/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraOutputOperator.java
+++ b/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraOutputOperator.java
@@ -46,7 +46,7 @@
private ArrayList<DataType> columnDataTypes;
private ArrayList<String> expressions;
private transient ArrayList<Object> getters;
-
+ private boolean isfirstTuple;
/*
* An ArrayList of Java expressions that will yield the field value from the POJO.
* Each expression corresponds to one column in the Cassandra table.
@@ -91,6 +91,7 @@
public CassandraOutputOperator()
{
super();
+ isfirstTuple = true;
columnDataTypes = new ArrayList<DataType>();
getters = new ArrayList<Object>();
}
@@ -99,9 +100,10 @@
@Override
public void processTuple(Object tuple)
{
- if (getters.isEmpty()) {
+ if (isfirstTuple) {
processFirstTuple(tuple);
}
+ isfirstTuple = false;
super.processTuple(tuple);
}
@@ -109,8 +111,10 @@
{
Class<?> fqcn = tuple.getClass();
int size = columnDataTypes.size();
+ LOG.debug("size is {}",size);
for (int i = 0; i < size; i++) {
DataType type = columnDataTypes.get(i);
+ LOG.debug("type is {}",type.getClass());
String getterExpression = expressions.get(i);
if (type.equals(DataType.Name.ASCII) || type.equals(DataType.Name.TEXT) || type.equals(DataType.Name.VARCHAR)) {
GetterString getVarchar = PojoUtils.createGetterString(fqcn, getterExpression);
@@ -144,6 +148,10 @@
GetterObject getObject = PojoUtils.createGetterObject(fqcn, getterExpression);
getters.add(getObject);
}
+ else
+ {
+ throw new UnsupportedOperationException("this operation is not supported"+type);
+ }
}
diff --git a/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java b/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
index e75a0ab..aec3e71 100644
--- a/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
+++ b/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
@@ -191,13 +191,13 @@
outputOperator.setup(context);
- List<TestEvent> events = Lists.newArrayList();
+ List<InnerObj> events = Lists.newArrayList();
for (int i = 0; i < 10; i++) {
- events.add(new TestEvent(i));
+ events.add(new InnerObj(i));
}
outputOperator.beginWindow(0);
- for (TestEvent event : events) {
+ for (InnerObj event : events) {
outputOperator.input.process(event);
}
outputOperator.endWindow();
@@ -205,7 +205,7 @@
Assert.assertEquals("rows in db", 10, outputOperator.getNumOfEventsInStore());
}
- @Test
+ //@Test
public void TestCassandraInputOperator()
{
CassandraStore store = new CassandraStore();
@@ -257,6 +257,11 @@
private int ID=11;
+ private InnerObj(int i)
+ {
+ ID = i;
+ }
+
/**
* @return the int ID
*/
diff --git a/library/src/main/java/com/datatorrent/lib/db/AbstractBatchTransactionableStoreOutputOperator.java b/library/src/main/java/com/datatorrent/lib/db/AbstractBatchTransactionableStoreOutputOperator.java
index ce05bbe..923a719 100644
--- a/library/src/main/java/com/datatorrent/lib/db/AbstractBatchTransactionableStoreOutputOperator.java
+++ b/library/src/main/java/com/datatorrent/lib/db/AbstractBatchTransactionableStoreOutputOperator.java
@@ -41,6 +41,7 @@
@Override
public void processTuple(T tuple)
{
+ System.out.println("in process tuple");
tuples.add(tuple);
}
@@ -54,6 +55,7 @@
/**
* Processes the whole batch at the end window and writes to the store.
*
+ * @param tuples
*/
public abstract void processBatch(Collection<T> tuples);
diff --git a/library/src/main/java/com/datatorrent/lib/db/AbstractTransactionableStoreOutputOperator.java b/library/src/main/java/com/datatorrent/lib/db/AbstractTransactionableStoreOutputOperator.java
index c3cf5b7..48ff88e 100644
--- a/library/src/main/java/com/datatorrent/lib/db/AbstractTransactionableStoreOutputOperator.java
+++ b/library/src/main/java/com/datatorrent/lib/db/AbstractTransactionableStoreOutputOperator.java
@@ -56,6 +56,9 @@
@Override
public void process(T t)
{
+ System.out.println("in process");
+ System.out.println("committedWindowId is " + committedWindowId);
+ System.out.println("currentWindowId is " + currentWindowId);
if (committedWindowId < currentWindowId) {
processTuple(t);
}
@@ -91,6 +94,7 @@
appId = context.getValue(DAG.APPLICATION_ID);
operatorId = context.getId();
committedWindowId = store.getCommittedWindowId(appId, operatorId);
+ System.out.println("committedWindowId is "+ committedWindowId);
}
catch (IOException ex) {
throw new RuntimeException(ex);
