Merge pull request #82 from s-ourabh/stickyDequeue-1-Support

Sticky dequeue 1 support

Author: ichokshi2109

clients/src/main/java/org/oracle/okafka/clients/consumer/KafkaConsumer.java

@@ -78,6 +78,7 @@
 import org.oracle.okafka.clients.CommonClientConfigs;
 import org.oracle.okafka.clients.Metadata;
 import org.oracle.okafka.clients.NetworkClient;
+import org.oracle.okafka.clients.TopicTeqParameters;
 import org.apache.kafka.clients.consumer.internals.ConsumerInterceptors;
 import org.oracle.okafka.clients.consumer.internals.ConsumerNetworkClient;
 import org.oracle.okafka.clients.consumer.internals.FetchMetricsRegistry;
@@ -1215,13 +1216,13 @@ record = new ConsumerRecord<>("", -1, -1, -1, TimestampType.NO_TIMESTAMP_TYPE, n
 					subscriptions.position(tp, new FetchPosition(record.offset(), Optional.empty(),
 							new LeaderAndEpoch(Optional.empty(), Optional.empty())));
 				} catch (IllegalStateException isE) {
-					if (metadata.getDBMajorVersion() < 23) {
+					TopicTeqParameters teqParam = metadata.topicParaMap.get(topic);
+					int stickyDeqParam = teqParam != null ? teqParam.getStickyDeq(): 2;
+					if (metadata.getDBMajorVersion() < 23 || stickyDeqParam == 1) {
 						// Partition assigned by TEQ Server not through JoinGroup/Sync
 						subscriptions.assignFromSubscribed(Collections.singleton(tp));
 						subscriptions.seek(tp, 0);
 						subscriptions.completeValidation(tp);
-						subscriptions.position(tp, new FetchPosition(record.offset(), Optional.empty(),
-								new LeaderAndEpoch(Optional.empty(), Optional.empty())));
 					}
 					subscriptions.position(tp, new FetchPosition(record.offset(), Optional.empty(),
 							new LeaderAndEpoch(Optional.empty(), Optional.empty())));

clients/src/main/java/org/oracle/okafka/clients/consumer/internals/ConsumerNetworkClient.java

@@ -63,6 +63,7 @@
 import org.oracle.okafka.clients.KafkaClient;
 import org.oracle.okafka.clients.Metadata;
 import org.oracle.okafka.clients.NetworkClient;
+import org.oracle.okafka.clients.TopicTeqParameters;
 import org.oracle.okafka.clients.consumer.KafkaConsumer.FetchManagerMetrics;
 import org.oracle.okafka.clients.consumer.internals.SubscriptionState.FetchPosition;
 import org.apache.kafka.clients.RequestCompletionHandler;
@@ -250,7 +251,9 @@ public void onComplete(ClientResponse response) {
 				Node node = poll.getKey();
 				log.debug("Fetch Records for topic " + poll.getValue() + " from host " + node );
 				String topic =  poll.getValue();
-				if(metadata.topicParaMap.get(topic).getStickyDeq() != 2) {
+				TopicTeqParameters teqParam = metadata.topicParaMap.get(topic);
+				int stickyDeqParam = teqParam != null ? teqParam.getStickyDeq(): 2;
+				if(stickyDeqParam == 0) {
 					String errMsg = "Topic " + topic + " is not an Oracle kafka topic, Please drop and re-create topic"
 							+" using Admin.createTopics() or dbms_aqadm.create_database_kafka_topic procedure";
 					throw new InvalidTopicException(errMsg);				
@@ -278,7 +281,7 @@ public void onComplete(ClientResponse response) {
 
 		return this.messages;
 	}
-	
+
 	/**
 	 * 
 	 * @return map of <node , topic> . Every node is leader for its corresponding topic.
@@ -397,7 +400,6 @@ private void joinGroupifNeeded(ClientResponse response, long timeoutMs) {
 			long elapsed = response.requestLatencyMs();
 			long prevTime = time.milliseconds();
 			long current;
-
 			while(elapsed < timeoutMs && rejoinNeeded(exception)) {
 				log.debug("JoinGroup Is Needed");
 				if (needsJoinPrepare) {
@@ -415,6 +417,7 @@ private void joinGroupifNeeded(ClientResponse response, long timeoutMs) {
 				elapsed = elapsed + (current - prevTime);
 				prevTime = current;
 			}
+			
 		} catch(Exception e)
 		{
 			log.error(e.getMessage(), e);
@@ -691,7 +694,6 @@ public boolean mayBeTriggerSubcription(long timeout) {
 
 		if(!subscriptions.subscription().equals(subscriptionSnapshot)) {
 			boolean noSubExist = false;
-			rejoin = true;
 			String topic = getSubscribableTopics();
 			long now = time.milliseconds();
 			Node node = client.leastLoadedNode(now);