|
7 | 7 |
|
8 | 8 | package org.oracle.okafka.examples; |
9 | 9 |
|
| 10 | +import java.io.FileNotFoundException; |
| 11 | +import java.io.IOException; |
| 12 | +import java.io.InputStream; |
10 | 13 | import java.util.Properties; |
11 | 14 | import java.sql.Connection; |
12 | 15 | import java.time.Duration; |
|
24 | 27 |
|
25 | 28 | import org.oracle.okafka.clients.consumer.KafkaConsumer; |
26 | 29 |
|
27 | | - |
28 | 30 | public class TransactionalConsumerOKafka { |
29 | 31 |
|
30 | | - // Dummy implementation of ConsumerRebalanceListener interface |
31 | | - // It only maintains the list of assigned partitions in assignedPartitions list |
32 | | - static class ConsumerRebalance implements ConsumerRebalanceListener { |
33 | | - |
34 | | - public List<TopicPartition> assignedPartitions = new ArrayList(); |
| 32 | + public static void main(String[] args) { |
| 33 | + System.setProperty("org.slf4j.simpleLogger.defaultLogLevel", "DEBUG"); |
35 | 34 |
|
36 | | - @Override |
37 | | - public synchronized void onPartitionsAssigned(Collection<TopicPartition> partitions) { |
38 | | - System.out.println("Newly Assigned Partitions:"); |
39 | | - for (TopicPartition tp :partitions ) { |
40 | | - System.out.println(tp); |
41 | | - assignedPartitions.add(tp); |
| 35 | + // Get application properties |
| 36 | + Properties appProperties = null; |
| 37 | + try { |
| 38 | + appProperties = getProperties(); |
| 39 | + if (appProperties == null) { |
| 40 | + System.out.println("Application properties not found!"); |
| 41 | + System.exit(-1); |
42 | 42 | } |
| 43 | + } catch (Exception e) { |
| 44 | + System.out.println("Application properties not found!"); |
| 45 | + System.out.println("Exception: " + e); |
| 46 | + System.exit(-1); |
43 | 47 | } |
44 | 48 |
|
45 | | - @Override |
46 | | - public synchronized void onPartitionsRevoked(Collection<TopicPartition> partitions) { |
47 | | - System.out.println("Revoked previously assigned partitions. "); |
48 | | - for (TopicPartition tp :assignedPartitions ) { |
49 | | - System.out.println(tp); |
50 | | - } |
51 | | - assignedPartitions.clear(); |
52 | | - } |
53 | | - } |
| 49 | + String topicName = appProperties.getProperty("topic.name", "TXEQ"); |
| 50 | + appProperties.remove("topic.name"); // Pass props to build OKafkaProducer |
54 | 51 |
|
55 | | - public static void main(String[] args) { |
56 | | - Properties props = new Properties(); |
57 | | - |
58 | | - // Option 1: Connect to Oracle Database with database username and password |
59 | | - props.put("security.protocol","PLAINTEXT"); |
60 | | - //IP or Host name where Oracle Database 23ai is running and Database Listener's Port |
61 | | - props.put("bootstrap.servers", "localhost:1521"); |
62 | | - props.put("oracle.service.name", "freepdb1"); //name of the service running on the database instance |
63 | | - // location for ojdbc.properties file where user and password properties are saved |
64 | | - props.put("oracle.net.tns_admin","."); |
65 | | - |
66 | | - /* |
67 | | - //Option 2: Connect to Oracle Autonomous Database using Oracle Wallet |
68 | | - //This option to be used when connecting to Oracle autonomous database instance on OracleCloud |
69 | | - props.put("security.protocol","SSL"); |
70 | | - // location for Oracle Wallet, tnsnames.ora file and ojdbc.properties file |
71 | | - props.put("oracle.net.tns_admin","."); |
72 | | - props.put("tns.alias","Oracle23ai_high"); |
73 | | - */ |
74 | | - |
75 | | - //Consumer Group Name |
76 | | - props.put("group.id" , "CG1"); |
77 | | - props.put("enable.auto.commit","false"); |
78 | | - |
79 | | - // Maximum number of records fetched in single poll call |
80 | | - props.put("max.poll.records", 10); |
81 | | - |
82 | | - props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); |
83 | | - props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); |
84 | | - |
85 | | - Consumer<String , String> consumer = new KafkaConsumer<String, String>(props); |
| 52 | + Consumer<String , String> consumer = new KafkaConsumer<String, String>(appProperties); |
86 | 53 | ConsumerRebalanceListener rebalanceListener = new ConsumerRebalance(); |
87 | 54 |
|
88 | | - consumer.subscribe(Arrays.asList("TXEQ"), rebalanceListener); |
| 55 | + consumer.subscribe(Arrays.asList(topicName), rebalanceListener); |
89 | 56 |
|
90 | 57 | int expectedMsgCnt = 100; |
91 | 58 | int msgCnt = 0; |
@@ -152,4 +119,55 @@ private static void processRecord(Connection conn, ConsumerRecord<String, String |
152 | 119 | { |
153 | 120 | //Application specific logic to process the message |
154 | 121 | } |
| 122 | + |
| 123 | + |
| 124 | + private static java.util.Properties getProperties() throws IOException { |
| 125 | + InputStream inputStream = null; |
| 126 | + Properties appProperties; |
| 127 | + |
| 128 | + try { |
| 129 | + Properties prop = new Properties(); |
| 130 | + String propFileName = "config.properties"; |
| 131 | + inputStream = TransactionalConsumerOKafka.class.getClassLoader().getResourceAsStream(propFileName); |
| 132 | + if (inputStream != null) { |
| 133 | + prop.load(inputStream); |
| 134 | + } else { |
| 135 | + throw new FileNotFoundException("property file '" + propFileName + "' not found."); |
| 136 | + } |
| 137 | + appProperties = prop; |
| 138 | + |
| 139 | + } catch (Exception e) { |
| 140 | + System.out.println("Exception: " + e); |
| 141 | + throw e; |
| 142 | + } finally { |
| 143 | + if (inputStream != null) |
| 144 | + inputStream.close(); |
| 145 | + } |
| 146 | + return appProperties; |
| 147 | + } |
| 148 | + |
| 149 | + // Dummy implementation of ConsumerRebalanceListener interface |
| 150 | + // It only maintains the list of assigned partitions in assignedPartitions list |
| 151 | + static class ConsumerRebalance implements ConsumerRebalanceListener { |
| 152 | + |
| 153 | + public List<TopicPartition> assignedPartitions = new ArrayList(); |
| 154 | + |
| 155 | + @Override |
| 156 | + public synchronized void onPartitionsAssigned(Collection<TopicPartition> partitions) { |
| 157 | + System.out.println("Newly Assigned Partitions:"); |
| 158 | + for (TopicPartition tp :partitions ) { |
| 159 | + System.out.println(tp); |
| 160 | + assignedPartitions.add(tp); |
| 161 | + } |
| 162 | + } |
| 163 | + |
| 164 | + @Override |
| 165 | + public synchronized void onPartitionsRevoked(Collection<TopicPartition> partitions) { |
| 166 | + System.out.println("Revoked previously assigned partitions. "); |
| 167 | + for (TopicPartition tp :assignedPartitions ) { |
| 168 | + System.out.println(tp); |
| 169 | + } |
| 170 | + assignedPartitions.clear(); |
| 171 | + } |
| 172 | + } |
155 | 173 | } |
0 commit comments