Changes done for v1.1.6.

Author: nysenthil

com.ibm.streamsx.websocket/CHANGELOG.md

@@ -1,5 +1,10 @@
 Changes
 =======
+## v1.1.6:
+* Oct/18/2023
+* Added new logic in the HttpPost operator to retry PUT, POST or GET operation in case of both for Java exceptions and for a user configured list of HTTP status codes such as 503, 408, 504 etc.
+* Added a HttpPost operator parameter to control which HTTP status codes will trigger retry: httpStatusCodesThatRequireRetry
+
 ## v1.1.5:
 * Oct/04/2023
 * Added new logic in the HttpPost operator to retry PUT, POST or GET operation in case of an error.

com.ibm.streamsx.websocket/impl/java/src/com/ibm/streamsx/websocket/op/HttpPost.java

@@ -9,7 +9,7 @@
 /*
 ==================================================================
 First created on: Mar/15/2020
-Last modified on: Oct/04/2023
+Last modified on: Oct/18/2023
 
 This Java operator is an utility operator available in the
 streamsx.websocket toolkit. It can be used to do HTTP(S) post of
@@ -56,9 +56,11 @@ TLS trust store and keystore creation (Java related) links:
 import java.security.cert.CertificateException;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.StringTokenizer;
 
 import javax.net.ssl.SSLContext;
 import org.apache.http.conn.ssl.NoopHostnameVerifier;
@@ -185,6 +187,9 @@ public class HttpPost extends AbstractOperator {
 	private int maxRetryAttempts = 0;
 	// Wait time is specified in Milliseconds
 	private int waitTimeBetweenRetry = 2000;
+	// Following instance variable will hold the HTTP status codes as 
+	// configured by the user for which we should trigger retry PUT, POST or GET.
+	private Set<Integer> httpStatusCodesSet = new HashSet<Integer>();
 
     // Create a HTTP put or post object.
 	// As of May/24/2020, this operator supports the posting of 
@@ -452,6 +457,9 @@ public final void process(StreamingInput<Tuple> inputStream, Tuple tuple) throws
     			// No retry attempt count was configured for this operator instance.
     			// So, no retry attempt needed and we can exit after just making the 
     			// intended PUT, POST or GET operation once.
+    			//
+    			// Set this operator metrics.
+    			nHttpPostFailed.increment();
     			break;
     		}
 
@@ -464,6 +472,9 @@ public final void process(StreamingInput<Tuple> inputStream, Tuple tuple) throws
     				". Giving up the HTTP " + httpMethod + 
     				" operation for the current incoming tuple after making " + 
     				maxRetryAttempts + " retry attempts.");
+
+    			// Set this operator metrics.
+    			nHttpPostFailed.increment();
     			break;
     		}
 
@@ -687,14 +698,6 @@ public final void process(StreamingInput<Tuple> inputStream, Tuple tuple) throws
 
 		        int responseStatusCode = response.getStatusLine().getStatusCode();
 
-		        // Update the operator metrics.
-		        if (responseStatusCode < 200 || responseStatusCode > 299) {
-		        	nHttpPostFailed.increment();
-		        } else {
-		        	nDataItemsSent.increment();
-		        	nDataBytesSent.setValue(dataBytesSent);
-		        }
-		        
 		        String responseStatusReason = response.getStatusLine().getReasonPhrase();
 		        // Let us now parse all the response headers and populate them in
 		        // a map to be sent in the output tuple later in the code below.
@@ -764,6 +767,28 @@ public final void process(StreamingInput<Tuple> inputStream, Tuple tuple) throws
 			        httpPost.releaseConnection();
 		        }
 
+		        // Before sending the output tuple, check if the HTTP status code that we received 
+		        // now is configured by the user to trigger a retry of PUT or POST operation.
+		        if(httpStatusCodesSet.contains(new Integer(responseStatusCode))) {
+	    			Logger.getLogger(this.getClass()).error("Operator=" + context.getName() + ", PE=" + 
+	        			context.getPE().getPEId() + ", Job=" + context.getPE().getJobId() +
+	        			", Incoming tuple number=" + httpPostCnt + 
+	        			". HTTP " + httpMethod + 
+	        			" operation for the current incoming tuple received a HTTP status code " +
+	        			responseStatusCode + " that must trigger a retry as configured by the user.");
+		        	// Continue the while loop to decide if we must make a 
+		        	// retry attempt after this executed PUT or POST operation.
+	    			continue;
+		        }
+
+		        // Update the operator metrics.
+		        if (responseStatusCode < 200 || responseStatusCode > 299) {
+		        	nHttpPostFailed.increment();
+		        } else {
+		        	nDataItemsSent.increment();
+		        	nDataBytesSent.setValue(dataBytesSent);
+		        }
+		        
 		        outTuple.setInt("statusCode", responseStatusCode);
 		        outTuple.setString("statusMessage", responseStatusReason);
 		        outTuple.setMap("responseHeaders", responseHeadersMap);
@@ -914,14 +939,6 @@ public final void process(StreamingInput<Tuple> inputStream, Tuple tuple) throws
 
 		        int responseStatusCode = response.getStatusLine().getStatusCode();
 
-		        // Update the operator metrics.
-		        if (responseStatusCode < 200 || responseStatusCode > 299) {
-		        	nHttpPostFailed.increment();
-		        } else {
-		        	nDataItemsSent.increment();
-		        	nDataBytesSent.setValue(dataBytesSent);
-		        }
-		        
 		        String responseStatusReason = response.getStatusLine().getReasonPhrase();
 		        // Let us now parse all the response headers and populate them in
 		        // a map to be sent in the output tuple later in the code below.
@@ -986,7 +1003,29 @@ public final void process(StreamingInput<Tuple> inputStream, Tuple tuple) throws
 		        // If we don't do this, it will start hanging when doing the 
 		        // next HTTP GET/PUT/POST for the next incoming tuple.
 			    httpGet.releaseConnection();
-		        
+
+		        // Before sending the output tuple, check if the HTTP status code that we received 
+		        // now is configured by the user to trigger a retry of GET operation.
+		        if(httpStatusCodesSet.contains(new Integer(responseStatusCode))) {
+	    			Logger.getLogger(this.getClass()).error("Operator=" + context.getName() + ", PE=" + 
+	        			context.getPE().getPEId() + ", Job=" + context.getPE().getJobId() +
+	        			", Incoming tuple number=" + httpPostCnt + 
+	        			". HTTP " + httpMethod + 
+	        			" operation for the current incoming tuple received a HTTP status code " +
+	        			responseStatusCode + " that must trigger a retry as configured by the user.");
+		        	// Continue the while loop to decide if we must make a 
+		        	// retry attempt after this executed GET operation.
+	    			continue;
+		        }
+
+		        // Update the operator metrics.
+		        if (responseStatusCode < 200 || responseStatusCode > 299) {
+		        	nHttpPostFailed.increment();
+		        } else {
+		        	nDataItemsSent.increment();
+		        	nDataBytesSent.setValue(dataBytesSent);
+		        }
+
 		        outTuple.setInt("statusCode", responseStatusCode);
 		        outTuple.setString("statusMessage", responseStatusReason);
 		        outTuple.setMap("responseHeaders", responseHeadersMap);
@@ -1154,7 +1193,7 @@ public void setMaxRetryAttempts(int val) {
     	if(val >= 0) {
     		maxRetryAttempts = val;
     	}
-    }
+    }	
 
     @Parameter (name="waitTimeBetweenRetry", 
 		description="This parameter specifies wait time in milliseconds between retry attempts after an error on a PUT, POST or GET operation for a given incoming tuple. Use this feature sparingly so as not to cause back pressure when this operator is processing a high volume of incoming tuples. (Default: 2000 milliseconds)", optional=true)
@@ -1164,6 +1203,26 @@ public void setWaitTimeBetweenRetry(int val) {
     		waitTimeBetweenRetry = val;
     	}
     }
+
+    @Parameter (name="httpStatusCodesThatRequireRetry",
+    	description="This parameter specifies a comma separated string containing zero or more HTTP status codes that must trigger a retry attempt on a PUT, POST or GET operation for a given incoming tuple. e-g: '503, 408, 504' (Default: Empty string)", optional=true)
+    public void setHttpStatusCodesThatRequireRetry(String val) {
+    	// We will parse this comma separated string containing zero or more
+    	// HTTP status codes and store them as elements in a set.
+    	if(val.length() > 0) {
+    		// Tokenize the given string using comma as a separator.
+    		StringTokenizer st = new StringTokenizer(val, ",");
+    		
+    		// Iterate through the available tokens and store them in the set as an Integer.
+    		while(st.hasMoreTokens()) {
+    			String token = st.nextToken();
+    			// Trim the leading and trailing space characters.
+    			token = token.trim();
+    			Integer httpStatusCode = Integer.parseInt(token);
+    			httpStatusCodesSet.add(httpStatusCode);
+    		}
+    	}
+    }
 
     public static final String DESC = "This operator posts/sends the incoming tuple's text or binary content to a " +
     		"HTTP or HTTPS persistent (Keep-Alive) or non-persistent endpoint specified in the operator parameter named url. " + 

com.ibm.streamsx.websocket/info.xml

@@ -4,7 +4,7 @@
  <info:identity>
    <info:name>com.ibm.streamsx.websocket</info:name>
    <info:description>Provides C++ WebSocket source, analytic and sink operators</info:description>
-   <info:version>1.1.5</info:version>
+   <info:version>1.1.6</info:version>
    <info:requiredProductVersion>4.2.1.0</info:requiredProductVersion>
  </info:identity>
  <info:dependencies/>

samples/HttpPostTester/com.ibm.streamsx.websocket.sample/HttpPostTester.spl

@@ -8,7 +8,7 @@
 /*
 ==================================================================
 First created on: Feb/13/2020
-Last modified on: Oct/04/2023
+Last modified on: Oct/18/2023
 
 This main composite file contains the logic to send
 a payload (plain text, JSON or XML) to a given HTTP(S) endpoint.
@@ -134,7 +134,17 @@ composite HttpPostTester {
 		// Do you want to wait between retry attempts for PUT, POST or GET operation?
 		expression<int32> $waitTimeBetweenRetry :
 			(int32)getSubmissionTimeValue("waitTimeBetweenRetry", "2000");		
-			
+
+		// Which HTTP status codes must trigger a retry operation?	
+		//	
+		// All Java exceptions will trigger retrying of the PUT, POST or GET operation when
+		// the max retry attempts parameter specified above is set to a non-zero value.
+		// However, if a retry is required for certain HTTP status codes, then the user 
+		// must give a comma separated string containing zero or more HTTP status codes.
+		// e-g: "503, 408, 504" 
+		expression<rstring> $httpStatusCodesThatRequireRetry :
+			getSubmissionTimeValue("httpStatusCodesThatRequireRetry", "");
+		
 		// What is the desired message rate we want to test with.
 		expression<float64> $maxMessageRate : 
 			(float64)getSubmissionTimeValue("MaxMessageRate", "500.00");
@@ -388,6 +398,15 @@ composite HttpPostTester {
 				//		
 				// Do you want to wait between retry attempts for PUT, POST or GET operation?
 				waitTimeBetweenRetry: $waitTimeBetweenRetry;
+				
+				// Which HTTP status codes must trigger a retry operation?
+				//		
+				// All Java exceptions will trigger retrying of the PUT, POST or GET operation when
+				// the max retry attempts parameter specified above is set to a non-zero value.
+				// However, if a retry is required for certain HTTP status codes, then the user 
+				// must give a comma separated string containing zero or more HTTP status codes.
+				// e-g: "503, 408, 504" 
+				httpStatusCodesThatRequireRetry : $httpStatusCodesThatRequireRetry;
 		}
 
 		// We can write the HTTP response somewhere or nowhere.
@@ -483,6 +502,15 @@ composite HttpPostTester {
 				//		
 				// Do you want to wait between retry attempts for PUT, POST or GET operation?
 				waitTimeBetweenRetry: $waitTimeBetweenRetry;
+				
+				// Which HTTP status codes must trigger a retry operation?	
+				//	
+				// All Java exceptions will trigger retrying of the PUT, POST or GET operation when
+				// the max retry attempts parameter specified above is set to a non-zero value.
+				// However, if a retry is required for certain HTTP status codes, then the user 
+				// must give a comma separated string containing zero or more HTTP status codes.
+				// e-g: "503, 408, 504" 
+				httpStatusCodesThatRequireRetry : $httpStatusCodesThatRequireRetry;
 		}
 
 		// Let us write the HTTP text based response received from the remote server to a file.

samples/HttpPostTester/etc/run-http-post-test.sh

@@ -2,7 +2,7 @@
 #
 #--------------------------------------------------------------------
 # First created on: Apr/23/2020
-# Last modified on: Oct/04/2023
+# Last modified on: Oct/18/2023
 #
 # This is a script that I used to test the HttpPost operator in
 # our IBM Streams lab in New York. You can make minor changes here and
@@ -30,4 +30,4 @@ sleep 10
 
 echo Starting the HttpPostTester.
 # Start a Java based HTTP POST client application on instance i2
-streamtool submitjob -d d1 -i i2 ~/workspace32/HttpPostTester/output/com.ibm.streamsx.websocket.sample.HttpPostTester.sab -P Url=http://b0513:8080/MyServices/Banking/Deposit -P NumSenders=1 -P createPersistentHttpConnection=true -P LogHttpPostActions=true -P MaxMessageRate=50.0 -P MessageBatchingTime=6.0 -P httpTimeout=30 -P delayBetweenConsecutiveHttpPosts=0 -P tlsAcceptAllCertificates=true -P maxRetryAttempts=5 -P waitTimeBetweenRetry=5000
+streamtool submitjob -d d1 -i i2 ~/workspace32/HttpPostTester/output/com.ibm.streamsx.websocket.sample.HttpPostTester.sab -P Url=http://b0513:8080/MyServices/Banking/Deposit -P NumSenders=1 -P createPersistentHttpConnection=true -P LogHttpPostActions=true -P MaxMessageRate=50.0 -P MessageBatchingTime=6.0 -P httpTimeout=30 -P delayBetweenConsecutiveHttpPosts=0 -P tlsAcceptAllCertificates=true -P maxRetryAttempts=5 -P waitTimeBetweenRetry=5000 -P httpStatusCodesThatRequireRetry="503, 408, 504"

samples/HttpPostTester/info.xml

@@ -3,13 +3,13 @@
   <info:identity>
     <info:name>HttpPostTester</info:name>
     <info:description>It is used for testing the HttpPost operator available in the streamsx.websocket toolkit.</info:description>
-    <info:version>1.0.5</info:version>
+    <info:version>1.0.6</info:version>
     <info:requiredProductVersion>4.2.1.0</info:requiredProductVersion>
   </info:identity>
   <info:dependencies>
     <info:toolkit>
       <common:name>com.ibm.streamsx.websocket</common:name>
-      <common:version>[1.0.7,7.0.0]</common:version>
+      <common:version>[1.1.6,2.0.0)</common:version>
     </info:toolkit>
   </info:dependencies>
-</info:toolkitInfoModel>
+</info:toolkitInfoModel>

websocket-tech-brief.txt

@@ -1,6 +1,6 @@
 ============================================================
 First created on: February/22/2020
-Last modified on: October/04/2023
+Last modified on: October/18/2023
 
 Purpose of this toolkit
 -----------------------