Client support for lightweight consumer #88
Conversation
oracle-contributor-agreement
bot
added
the
OCA Verified
| commitSyncAll(node, topic, partitions, priorities, subshards, sequences); | ||
| } | ||
|
|
||
| public void createLightWeightSub(String topic, Node node) throws SQLException { |
There was a problem hiding this comment.
createLightWeightSub where do we invoke this from ?
Also, if we are creating a lightweight sub, can we set a flag in TopicConsumers to indicate that this is a lightweight sub.
|
|
||
| Connection con = ((AQjmsSession)topicConsumersMap.get(node).getSession()).getDBConnection(); | ||
| String user = con.getMetaData().getUserName(); | ||
| cStmt = con.prepareCall("{call dbms_teqk.AQ$_COMMITSYNC(?, ?, ?, ?, ?, ?, ?)}"); |
There was a problem hiding this comment.
We should be able to cache the callable statements and create it once only. We can close them from consumer.close()
| } | ||
|
|
||
| private String getCurrentUser(Node node) throws SQLException, JMSException { | ||
| Connection con = ((AQjmsSession) topicConsumersMap.get(node).getSession()).getDBConnection(); |
There was a problem hiding this comment.
See if getCurrentUser() can use getConnection() ?
| } | ||
| } | ||
|
|
||
| public void lightWeightSeektoBeginning(Node node, String topic, int[] partition_id, int[] priority) { |
There was a problem hiding this comment.
We need to modify the seek() method to use lightweight seek.
We need to invoke lightWeightSeektoBeginning when OffsetRessetStrategy is set to SEEK_TO_BEGINNING
| @@ -58,6 +58,9 @@ public class CommonClientConfigs extends org.apache.kafka.clients.CommonClientCo | |||
|
|
|||
| public static final String ORACLE_TRANSACTIONAL_PRODUCER ="oracle.transactional.producer"; | |||
|
|
|||
| public static final String ORACLE_CONSUMER_LIGHTWEIGHT = "oracle.consumer.lightweight"; | |||
| public static final String ORACLE_CONSUMER_LIGHTWEIGHT_DOC = "Creates a light weight subscriber"; | |||
There was a problem hiding this comment.
why do we want that in CommonClientConfigs and ConsumerConfig ?
| } | ||
| } | ||
|
|
||
| public void closeCallableStmt(Node node) { |
There was a problem hiding this comment.
invoke this when we close the consumer
| private void commitOffsetsLightWeightSub(Node node, String topic, Map<TopicPartition, OffsetAndMetadata> offsets) { | ||
|
|
||
| final int OFFSET_DIVISOR = 20000; | ||
| int size = offsets.size(); |
There was a problem hiding this comment.
check for an existing variable for SUBSHARDSIZE
| } | ||
| else { | ||
| long subshard = offset / 20000; | ||
| long sequence = offset % 20000; |
There was a problem hiding this comment.
Replace 20000 with appropriate variable
| cStmt.execute(); | ||
| log.debug("lightWeightSeektoEnd executed for topic: {}, partitions: {}", topic, partition_id.length); | ||
| } catch (Exception ex) { | ||
| log.error("Error in lightWeightSeektoEnd for topic: " + topic + ", node: " + node, ex); |
There was a problem hiding this comment.
throw the exception to caller as well. That will be used to set in the response.
| @@ -1586,9 +1846,11 @@ public TopicConsumers(Node node,int mode) throws JMSException { | |||
|
|
|||
| try { | |||
| this.dbVersion = ConnectionUtils.getDBVersion(oConn); | |||
| this.lightWeightSub = configs.getBoolean(ConsumerConfig.ORACLE_CONSUMER_LIGHTWEIGHT); | |||
There was a problem hiding this comment.
get DBMajorVersion and if >= 26 and if LIGHTWEIGHT property is set then only allow to crate LTWT subscriber
added createLightWeightSub(), commit() methods and seek() methods for kafka lightweight subscriber