wip examples 23.4 fixed consumer

Author: pasimoes

README.md

@@ -22,7 +22,8 @@ To run `OKafka application` against Oracle Database, a database user must be cre
 
 ```roomsql
 create user <user> identified by <password>
-GRANT CONNECT , RESOURCE to user;
+GRANT AQ_USER_ROLE to user;
+GRANT CONNECT, RESOURCE, unlimited tablespace to user;
 GRANT EXECUTE on DBMS_AQ to user;
 GRANT EXECUTE on DBMS_AQADM to user;
 GRANT EXECUTE on DBMS_AQIN to user;
@@ -33,7 +34,8 @@ GRANT SELECT on GV_$INSTANCE to user;
 GRANT SELECT on GV_$LISTENER_NETWORK to user;
 GRANT SELECT on GV_$PDBS to user;
 GRANT SELECT on USER_QUEUE_PARTITION_ASSIGNMENT_TABLE to user;
-exec DBMS_AQADM.GRANT_PRIV_FOR_RM_PLAN('user');
+GRANT SELECT on SYS.DBA_RSRC_PLAN_DIRECTIVES to user;
+EXEC DBMS_AQADM.GRANT_PRIV_FOR_RM_PLAN('user');
 ```
 
 Once user is created and above privileges are granted, connect to Oracle Database as this user and create a Transactional Event Queue using below PL/SQL script. One can also use `KafkaAdmin` interface as shown in `CreateTopic.java` in `examples` directory to create a Transactional Event Queue. 

examples/consumer/src/main/java/org/oracle/okafka/examples/ConsumerOKafka.java

@@ -21,7 +21,7 @@
 
 public class ConsumerOKafka {
 	public static void main(String[] args) {
-		System.setProperty("org.slf4j.simpleLogger.defaultLogLevel", "DEBUG");
+		System.setProperty("org.slf4j.simpleLogger.defaultLogLevel", "INFO");
 
 		// Get application properties
 		Properties appProperties = null;
@@ -43,20 +43,21 @@ public static void main(String[] args) {
 		KafkaConsumer<String , String> consumer = new KafkaConsumer<>(appProperties);
 		consumer.subscribe(Arrays.asList(topic));
 
-		while(true) {
+
 			try {
-				ConsumerRecords <String, String> records = consumer.poll(Duration.ofMillis(10000));
+				while(true) {
+					ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));
 
-				for (ConsumerRecord<String, String> record : records)
-					System.out.printf("partition = %d, offset = %d, key = %d, value =%s\n  ", record.partition(), record.offset(), record.key(), record.value());
+					for (ConsumerRecord<String, String> record : records)
+						System.out.printf("partition = %d, offset = %d, key = %s, value =%s\n  ", record.partition(), record.offset(), record.key(), record.value());
 
-				if(records != null && records.count() > 0) {
-					System.out.println("Committing records" + records.count());
-					consumer.commitSync();     
-				}
-				else {
-					System.out.println("No Record Fetched. Retrying in 1 second");
-					Thread.sleep(1000);
+					if (records != null && records.count() > 0) {
+						System.out.println("Committing records" + records.count());
+						consumer.commitSync();
+					} else {
+						System.out.println("No Record Fetched. Retrying in 1 second");
+						Thread.sleep(1000);
+					}
 				}
 			}catch(Exception e)
 			{
@@ -66,7 +67,7 @@ public static void main(String[] args) {
 			finally {
 				consumer.close();
 			}
-		}
+
 	}
 
 	private static java.util.Properties getProperties()  throws IOException {

examples/consumer/src/main/resources/config.properties

@@ -5,7 +5,6 @@
 #bootstrap.servers=<server address:server port>
 #oracle.service.name=<oracle database service>
 #oracle.net.tns_admin=<location of ojdbc.properties file>
-security.protocol="PLAINTEXT"
 bootstrap.servers=84.235.172.160:1521
 oracle.service.name="freepdb1"
 oracle.net.tns_admin=/Users/pasimoes/Work/Oracle/Code/okafka/23.4/okafka/examples

examples/producer/src/main/java/org/oracle/okafka/examples/ProducerOKafka.java

@@ -8,6 +8,7 @@
 package org.oracle.okafka.examples;
 
 import org.oracle.okafka.clients.producer.KafkaProducer;
+
 import org.apache.kafka.common.header.internals.RecordHeader;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;