Merge pull request #115 from IBMStreams/develop

streamsx.hbase merger develop to master

Author: anouri

com.ibm.streamsx.hbase/.classpath

@@ -18,7 +18,10 @@
 	<classpathentry kind="lib" path="opt/downloaded/hbase-client-1.4.9.jar"/>
 	<classpathentry kind="lib" path="opt/downloaded/hbase-common-1.4.9.jar"/>
 	<classpathentry kind="lib" path="opt/downloaded/hbase-protocol-1.4.9.jar"/>
+	<classpathentry kind="lib" path="opt/downloaded/htrace-core-3.1.0-incubating.jar"/>
 	<classpathentry kind="lib" path="opt/downloaded/htrace-core4-4.2.0-incubating.jar"/>
+	<classpathentry kind="lib" path="opt/downloaded/jackson-core-asl-1.9.13.jar"/>
+	<classpathentry kind="lib" path="opt/downloaded/jackson-mapper-asl-1.9.13.jar"/>
 	<classpathentry kind="lib" path="opt/downloaded/javaee-api-7.0.jar"/>
 	<classpathentry kind="lib" path="opt/downloaded/log4j-1.2.15.jar"/>
 	<classpathentry kind="lib" path="opt/downloaded/metrics-core-2.2.0.jar"/>
@@ -27,6 +30,6 @@
 	<classpathentry kind="lib" path="opt/downloaded/servlet-api-2.5.jar"/>
 	<classpathentry kind="lib" path="opt/downloaded/slf4j-api-1.7.10.jar"/>
 	<classpathentry kind="lib" path="opt/downloaded/slf4j-log4j12-1.7.10.jar"/>
-	<classpathentry kind="lib" path="opt/downloaded/zookeeper-3.4.13.jar"/>
+	<classpathentry kind="lib" path="opt/downloaded/zookeeper-3.4.6.jar"/>
 	<classpathentry kind="output" path="bin"/>
 </classpath>

com.ibm.streamsx.hbase/impl/java/src/com/ibm/streamsx/hbase/HBASEOperatorWithInput.java

@@ -24,17 +24,23 @@ public abstract class HBASEOperatorWithInput extends HBASEOperator {
 	protected int rowAttrIndex = -1;
 	protected int colFamilyIndex = -1;
 	protected int colQualifierIndex = -1;
+	protected int timestampAttributeIndex = -1;
 
 	protected MetaType rowAttrType = null;
 	protected int valueAttrIndex = -1;
 	protected MetaType valueAttrType = null;
 
-	protected MetaType colQualifierType = null, colFamilyType = null;
+	protected MetaType colQualifierType = null;
+	protected MetaType colFamilyType = null;
+	protected MetaType timestampType =null;
 
 	static final String COL_FAM_PARAM_NAME = "columnFamilyAttrName";
 	static final String COL_QUAL_PARAM_NAME = "columnQualifierAttrName";
 	static final String TABLE_PARAM_NAME = "tableName";
 	static final String ROW_PARAM_NAME = "rowAttrName";
+	static final String TIMESTAMP = "Timestamp";
+	static final String TIMESTAMP_ATTR = "TimestampAttrName";
+
 	byte colFamBytes[] = null;
 	byte colQualBytes[] = null;
 
@@ -98,12 +104,20 @@ protected byte[] getColumnQualifier(Tuple tuple) throws Exception {
 		}
 	}
 
+	protected long getTimeStamp(Tuple tuple) throws Exception {
+		if (timestampAttributeIndex > 0)
+		{
+			return tuple.getLong(timestampAttributeIndex);
+		}else{
+			return 0;
+		}
+	}
+
 	protected byte[] getValue(Tuple tuple) throws Exception {
 
 			return getBytes(tuple, valueAttrIndex, valueAttrType);
 	}
 
-	
 	/**
 	 * For {rowAttrName,columnFamilyAttrName,columnQualifierAttrName}, if
 	 * specified, ensures the attribute exists, and stores the index in class

com.ibm.streamsx.hbase/impl/java/src/com/ibm/streamsx/hbase/HBASEPut.java

@@ -7,6 +7,7 @@
 import java.util.ArrayList;
 import java.util.List;
 
+
 import org.apache.hadoop.hbase.TableNotFoundException;
 import org.apache.hadoop.hbase.client.BufferedMutator;
 import org.apache.hadoop.hbase.client.Put;
@@ -21,6 +22,7 @@
 import com.ibm.streams.operator.StreamingInput;
 import com.ibm.streams.operator.StreamingOutput;
 import com.ibm.streams.operator.Tuple;
+import com.ibm.streams.operator.TupleAttribute;
 import com.ibm.streams.operator.Type.MetaType;
 import com.ibm.streams.operator.compile.OperatorContextChecker;
 import com.ibm.streams.operator.meta.TupleType;
@@ -126,6 +128,8 @@ private enum PutMode {
 	private PutMode putMode = null;
 	protected String valueAttr = null;
 	final static String VALUE_NAME = "valueAttrName";
+	protected long timeStamp = 0;
+	protected String timestampAttribute = null; 
 	protected byte[][] qualifierArray = null;
 	protected MetaType[] attrType = null;
 	protected boolean bufferTransactions = false;
@@ -155,6 +159,17 @@ public void setValueAttr(String val) {
 		valueAttr = val;
 	}
 
+	@Parameter(name = TIMESTAMP, optional = true, description = "This parameter specifies the timestamp in milliseconds (INT64). The timestamp allows for versioning of the cells. Everytime HBaes make a PUT on a table it set the timestamp. By default this is the current time in milliseconds, but you can set your own timestamp as well with this parametr. Cannot be used with TimestampAttrName")
+	public void setTimestamp(long ts) {
+		timeStamp = ts;
+	}
+
+	@Parameter(name = TIMESTAMP_ATTR, optional = true, description = "Name of the attribute on the input tuple containing the timestamp in milliseconds. Cannot be used with Timestamp.")
+	public void setTimestampAtrr(String ts) {
+		timestampAttribute = ts;
+	}
+
+	
 	/**
 	 * Do any necessary compile time checks. It calls the checker of the super
 	 * class.
@@ -184,7 +199,9 @@ public static void checkForBatchSizeParam(OperatorContextChecker checker) {
 		} else if (context.getParameterNames().contains(BATCHSIZE_NAME)) {
 			System.err.println("The " + BATCHSIZE_NAME + " has been deprecated and should not be used.  Use the " +BUFFER_PARAM  + " parameter instead.");
 		}
-		
+		if ((!checker.checkExcludedParameters(TIMESTAMP,TIMESTAMP_ATTR))){
+		 	checker.setInvalidContext("The " + TIMESTAMP + " can not be used with " + TIMESTAMP_ATTR , null);
+		}
 	}
 
 	Logger logger = Logger.getLogger(this.getClass());
@@ -247,6 +264,17 @@ public synchronized void initialize(OperatorContext context)
 			valueAttrType = attr.getType().getMetaType();
 			putMode = PutMode.ENTRY;
 		}
+		
+		if (timestampAttribute != null) {
+			timestampAttributeIndex = checkAndGetIndex(schema, timestampAttribute);
+			timestampType = schema.getAttribute(timestampAttributeIndex).getType().getMetaType();
+			if (timestampType != MetaType.INT64){
+					String message = Messages.getString("HBASE_OP_INVALID_ATTR", timestampAttribute, timestampType);
+					Logger.getLogger(this.getClass()).trace(message);
+					System.err.print(message);
+			}
+		}
+
 	}
 
 	/**
@@ -284,7 +312,18 @@ public void process(StreamingInput<Tuple> stream, Tuple tuple)
 				case ENTRY:
 					byte colQ[] = getColumnQualifier(tuple);
 					byte value[] = getBytes(tuple, valueAttrIndex, valueAttrType);
-					myPut.addColumn(colF, colQ, value);
+					if (timestampAttribute != null) {
+						timeStamp = getTimeStamp(tuple);
+					}
+					
+					if (timeStamp > 0){
+						myPut.addColumn(colF, colQ, timeStamp, value);
+				        // System.out.println("timeStamp = " + timeStamp);					
+					}
+					else {
+						myPut.addColumn(colF, colQ, value);						
+					}
+					
 					break;
 				case RECORD:
 					Tuple values = tuple.getTuple(valueAttr);
@@ -308,8 +347,7 @@ public void process(StreamingInput<Tuple> stream, Tuple tuple)
 						byte checkColF[] = getBytes(checkTuple, checkColFIndex, checkColFType);
 						byte checkColQ[] = getBytes(checkTuple, checkColQIndex, checkColQType);
 						byte checkValue[] = getCheckValue(checkTuple);
-			
-						success = myTable.checkAndPut(checkRow, checkColF, checkColQ, checkValue, myPut);
+						success = myTable.checkAndPut(checkRow, checkColF, checkColQ, checkValue, myPut);																		
 					}else{
 						// set the success value without checkTuple
 						byte checkColQ[] = getColumnQualifier(tuple);
@@ -341,6 +379,7 @@ public void process(StreamingInput<Tuple> stream, Tuple tuple)
 				// Checks to see if an output tuple is necessary, and if so,
 				// submits it.
 				submitOutputTuple(tuple, success);
+				success = false;
 			} catch (Exception e) {
 				logger.error(e.getMessage());
 				submitErrorMessagee(e.getMessage(), tuple);
@@ -455,4 +494,4 @@ public void processPunctuation(StreamingInput<Tuple> stream,
 
 
 
-}
+}

com.ibm.streamsx.hbase/impl/java/src/com/ibm/streamsx/hbase/HBASEPutDelete.java

@@ -43,9 +43,10 @@ public abstract class HBASEPutDelete extends HBASEOperatorWithInput implements
 	protected int checkColFIndex = -1;
 	protected int checkColQIndex = -1;
 	protected int checkValueIndex = -1;
+	protected int checkTimestampIndex = -1;
 
 	protected MetaType checkColFType = null, checkColQType = null,
-			checkValueType = null;
+			checkValueType = null, checkTimestampType = null;
 
 	Logger logger = Logger.getLogger(this.getClass());
 
@@ -132,6 +133,13 @@ protected void establishCheckAttrMatching(Attribute checkAttr)
 			checkValueType = checkSchema.getAttribute(checkValueIndex)
 					.getType().getMetaType();
 		}
+
+		if (checkSchema.getAttribute("Timestamp") != null) {
+			checkTimestampIndex = checkAndGetIndex(checkSchema, "Timestamp");
+			checkTimestampType = checkSchema.getAttribute(checkTimestampIndex)
+					.getType().getMetaType();
+		}
+
 	}
 
 	byte[] getCheckValue(Tuple tuple) throws Exception {
@@ -140,6 +148,14 @@ byte[] getCheckValue(Tuple tuple) throws Exception {
 		} else
 			return null;
 	}
+	
+	byte[] getCheckTimeStamp(Tuple tuple) throws Exception {
+		if (checkTimestampIndex > 0) {
+			return getBytes(tuple, checkTimestampIndex, checkTimestampType);
+		} else
+			return null;
+	}
+	
 
 	/**
 	 * This checks that

com.ibm.streamsx.hbase/info.xml

@@ -184,9 +184,16 @@ Here is an example to connect to the HBase server with Kerberos Authentication:
   The first attribute in the optional error output port must be a 'rstring'.
   The second attribute in the error output port is optional and must be a 'TUPLE'.
 
+++ What is new in version 3.7.0
+
+* The HBASEPut operator provides 2 new parameters to add time stamp:  
+**Timestamp** This parameter specifies the timestamp in milliseconds (INT64). The timestamp allows for versioning of the cells. Every time HBaes make a PUT on a table it set the timestamp. By default this is the current time in milliseconds, but you can set your own timestamp as well with this parameter. Cannot be used with TimestampAttrName")
+**TimestampAttrName** Name of the attribute on the input tuple containing the timestamp in milliseconds. Cannot be used with Timestamp.
+
+The jar library zookeeper-3.4.13.jar has been replaced with **zookeeper-3.4.6.jar** .
 
 </info:description>
-    <info:version>3.6.0</info:version>
+    <info:version>3.7.0</info:version>
     <info:requiredProductVersion>4.0.0.0</info:requiredProductVersion>
   </info:identity>
   <info:dependencies/>

com.ibm.streamsx.hbase/pom.xml

@@ -15,7 +15,7 @@
         <groupId>com.ibm.streamsx.hbase</groupId>
         <artifactId>streamsx.hbase</artifactId>
         <packaging>jar</packaging>
-        <version>3.6.0</version>
+        <version>3.7.0</version>
         <name>com.ibm.streamsx.hbase</name>
         <repositories>
                 <repository>
@@ -337,7 +337,7 @@
                 <dependency>
                         <groupId>org.apache.zookeeper</groupId>
                         <artifactId>zookeeper</artifactId>
-                        <version>3.4.13</version>
+                        <version>3.4.6</version>
                         <exclusions>
                                 <exclusion>
                                         <groupId>*</groupId>