[WIP] Add data publishing capabilities to WSO2 OB IS Accelerator modules #832
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
![wso2-engineering[bot]](https://avatars.githubusercontent.com/in/1856828?s=60&v=4){kind=small}
Comment on lines
143
to
146
| buildFSEventExecutors(); | ||
| } catch (IOException | XMLStreamException | OMException e) { | ||
| throw new FinancialServicesRuntimeException("Error occurred while building configuration from " + | ||
| "financial-services.xml", e); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 1
Suggested change
| buildFSEventExecutors(); | |
| } catch (IOException | XMLStreamException | OMException e) { | |
| throw new FinancialServicesRuntimeException("Error occurred while building configuration from " + | |
| "financial-services.xml", e); | |
| } catch (IOException | XMLStreamException | OMException e) { | |
| log.error("Error occurred while building configuration from financial-services.xml", e.getMessage()); | |
| throw new FinancialServicesRuntimeException("Error occurred while building configuration from " + | |
| "financial-services.xml", e); |
Comment on lines
213
to
214
| private void buildFSEventExecutors() { | ||
|
|
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 2
Suggested change
| private void buildFSEventExecutors() { | |
| private void buildFSEventExecutors() { | |
| log.debug("Building Financial Services Event Executors configuration"); |
Comment on lines
237
to
241
|
|
||
| if (StringUtils.isEmpty(obExecutorClass)) { | ||
| //Throwing exceptions since we cannot proceed without invalid executor names | ||
| throw new FinancialServicesRuntimeException("Event Executor class is not defined " + | ||
| "correctly in open-banking.xml"); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 3
Suggested change
| if (StringUtils.isEmpty(obExecutorClass)) { | |
| //Throwing exceptions since we cannot proceed without invalid executor names | |
| throw new FinancialServicesRuntimeException("Event Executor class is not defined " + | |
| "correctly in open-banking.xml"); | |
| if (StringUtils.isEmpty(obExecutorClass)) { | |
| log.error("Event Executor class is not defined correctly in open-banking.xml"); | |
| //Throwing exceptions since we cannot proceed without invalid executor names | |
| throw new FinancialServicesRuntimeException("Event Executor class is not defined " + | |
| "correctly in open-banking.xml"); |
Comment on lines
+246
to
+247
| } | ||
| fsEventExecutors.put(priority, obExecutorClass); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 4
Suggested change
| } | |
| fsEventExecutors.put(priority, obExecutorClass); | |
| } | |
| fsEventExecutors.put(priority, obExecutorClass); | |
| log.debug("Added event executor: {} with priority: {}", obExecutorClass, priority); |
Comment on lines
317
to
318
| protected void buildDataPublishingStreams() { | ||
|
|
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 5
Suggested change
| protected void buildDataPublishingStreams() { | |
| protected void buildDataPublishingStreams() { | |
| log.debug("Building Data Publishing Streams configuration"); |
Comment on lines
352
to
357
|
|
||
| if (StringUtils.isEmpty(attributeName)) { | ||
| //Throwing exceptions since we cannot proceed without valid attribute names | ||
| throw new FinancialServicesRuntimeException( | ||
| "Data publishing attribute name is not defined " + | ||
| "correctly in financial-services.xml"); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 6
Suggested change
| if (StringUtils.isEmpty(attributeName)) { | |
| //Throwing exceptions since we cannot proceed without valid attribute names | |
| throw new FinancialServicesRuntimeException( | |
| "Data publishing attribute name is not defined " + | |
| "correctly in financial-services.xml"); | |
| if (StringUtils.isEmpty(attributeName)) { | |
| log.error("Data publishing attribute name is not defined correctly in financial-services.xml"); | |
| //Throwing exceptions since we cannot proceed without valid attribute names | |
| throw new FinancialServicesRuntimeException( | |
| "Data publishing attribute name is not defined " + | |
| "correctly in financial-services.xml"); |
Comment on lines
+378
to
+379
| String attributeKey = dataStreamName + "_" + attributeName; | ||
| dataPublishingValidationMap.put(attributeKey, metadata); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 7
Suggested change
| String attributeKey = dataStreamName + "_" + attributeName; | |
| dataPublishingValidationMap.put(attributeKey, metadata); | |
| } | |
| } | |
| if (log.isDebugEnabled()) { | |
| log.debug("Added data publishing stream: {} with {} attributes", dataStreamName, attributes.size()); | |
| } |
Comment on lines
27
to
30
|
|
||
| @Override | ||
| public void processEvent(FSEvent fsEvent) { | ||
|
|
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 8
Suggested change
| @Override | |
| public void processEvent(FSEvent fsEvent) { | |
| @Override | |
| public void processEvent(FSEvent fsEvent) { | |
| log.debug("Processing FSEvent: {}", fsEvent); | |
| } |
Comment on lines
42
to
47
|
|
||
| public FSEventQueue(int queueSize, int workerThreadCount) { | ||
|
|
||
| // Note : Using a fixed worker thread pool and a bounded queue to control the load on the server | ||
| executorService = Executors.newFixedThreadPool(workerThreadCount); | ||
| eventQueue = new ArrayBlockingQueue<>(queueSize); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 9
Suggested change
| public FSEventQueue(int queueSize, int workerThreadCount) { | |
| // Note : Using a fixed worker thread pool and a bounded queue to control the load on the server | |
| executorService = Executors.newFixedThreadPool(workerThreadCount); | |
| eventQueue = new ArrayBlockingQueue<>(queueSize); | |
| public FSEventQueue(int queueSize, int workerThreadCount) { | |
| log.info("Initializing FSEventQueue with queue size: " + queueSize + " and worker threads: " + workerThreadCount); | |
| // Note : Using a fixed worker thread pool and a bounded queue to control the load on the server | |
| executorService = Executors.newFixedThreadPool(workerThreadCount); | |
| eventQueue = new ArrayBlockingQueue<>(queueSize); |
Comment on lines
49
to
56
|
|
||
| public void put(FSEvent obEvent) { | ||
|
|
||
| try { | ||
| if (eventQueue.offer(obEvent)) { | ||
| executorService.submit(new FSQueueWorker(eventQueue, executorService)); | ||
| } else { | ||
| log.error("Event queue is full. Starting to drop events."); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 10
Suggested change
| public void put(FSEvent obEvent) { | |
| try { | |
| if (eventQueue.offer(obEvent)) { | |
| executorService.submit(new FSQueueWorker(eventQueue, executorService)); | |
| } else { | |
| log.error("Event queue is full. Starting to drop events."); | |
| public void put(FSEvent obEvent) { | |
| if (log.isDebugEnabled()) { | |
| log.debug("Attempting to add event to queue"); | |
| } | |
| try { | |
| if (eventQueue.offer(obEvent)) { | |
| executorService.submit(new FSQueueWorker(eventQueue, executorService)); | |
| } else { | |
| log.error("Event queue is full. Starting to drop events."); |
Comment on lines
58
to
65
| } catch (RejectedExecutionException e) { | ||
| log.warn("Task submission failed. Task queue might be full", e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| protected void finalize() throws Throwable { | ||
| executorService.shutdown(); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 11
Suggested change
| } catch (RejectedExecutionException e) { | |
| log.warn("Task submission failed. Task queue might be full", e); | |
| } | |
| } | |
| @Override | |
| protected void finalize() throws Throwable { | |
| executorService.shutdown(); | |
| log.warn("Task submission failed. Task queue might be full", e); | |
| } | |
| } | |
| @Override | |
| protected void finalize() throws Throwable { | |
| log.info("Shutting down FSEventQueue executor service"); | |
| executorService.shutdown(); | |
| super.finalize(); |
| this.executorService = executorService; | ||
| } | ||
|
|
||
| @Override |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 12
Suggested change
| @Override | |
| @Override | |
| public void run() { | |
| log.info("FSQueueWorker started processing events"); |
Comment on lines
+55
to
+56
| do { | ||
| FSEvent event = eventQueue.poll(); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 13
Suggested change
| do { | |
| FSEvent event = eventQueue.poll(); | |
| FSEvent event = eventQueue.poll(); | |
| if (event != null) { | |
| if (log.isDebugEnabled()) { | |
| log.debug("Processing FSEvent from queue"); | |
| } |
Comment on lines
+62
to
+63
| .getClassInstanceFromFQN(fsEventExecutors.get(integer))).collect(Collectors.toList()); | ||
| for (FSEventExecutor obEventExecutor : executorList) { |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 14
Suggested change
| .getClassInstanceFromFQN(fsEventExecutors.get(integer))).collect(Collectors.toList()); | |
| for (FSEventExecutor obEventExecutor : executorList) { | |
| for (FSEventExecutor obEventExecutor : executorList) { | |
| if (log.isDebugEnabled()) { | |
| log.debug("Executing event with executor: " + obEventExecutor.getClass().getSimpleName()); | |
| } | |
| obEventExecutor.processEvent(event); |
Comment on lines
+65
to
+66
| } | ||
| } else { |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 15
Suggested change
| } | |
| } else { | |
| } else { | |
| log.warn("Polled null event from queue, skipping processing"); |
Member
Author
There was a problem hiding this comment.
Already there's an error log
Comment on lines
30
to
34
|
|
||
| public FSEvent(String eventType, Map<String, Object> eventData) { | ||
|
|
||
| this.eventType = eventType; | ||
| this.eventData = eventData; |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 16
Suggested change
| public FSEvent(String eventType, Map<String, Object> eventData) { | |
| this.eventType = eventType; | |
| this.eventData = eventData; | |
| public FSEvent(String eventType, Map<String, Object> eventData) { | |
| log.debug("Creating FSEvent with type: {}", eventType); | |
| this.eventType = eventType; | |
| this.eventData = eventData; | |
| } |
| private PoolingHttpClientConnectionManager connectionManager; | ||
| private FSEventQueue fsEventQueue; | ||
| private Map<Integer, String> fsEventExecutors; | ||
|
|
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 17
Suggested change
| private FinancialServicesCommonDataHolder() { | |
| log.info("Initializing FinancialServicesCommonDataHolder"); |
| int workerThreadCount = | ||
| Integer.parseInt((String) FinancialServicesConfigParser.getInstance().getConfiguration() | ||
| .get(FinancialServicesConstants.EVENT_WORKER_THREAD_COUNT)); | ||
| fsEventQueue = new FSEventQueue(queueSize, workerThreadCount); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 18
Suggested change
| fsEventQueue = new FSEventQueue(queueSize, workerThreadCount); | |
| fsEventQueue = new FSEventQueue(queueSize, workerThreadCount); | |
| log.info("Initialized FSEventQueue with queue size: " + queueSize + " and worker thread count: " + workerThreadCount); |
| Integer.parseInt((String) FinancialServicesConfigParser.getInstance().getConfiguration() | ||
| .get(FinancialServicesConstants.EVENT_WORKER_THREAD_COUNT)); | ||
| fsEventQueue = new FSEventQueue(queueSize, workerThreadCount); | ||
| fsEventExecutors = FinancialServicesConfigParser.getInstance().getFinancialServicesEventExecutors(); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 19
Suggested change
| fsEventExecutors = FinancialServicesConfigParser.getInstance().getFinancialServicesEventExecutors(); | |
| fsEventExecutors = FinancialServicesConfigParser.getInstance().getFinancialServicesEventExecutors(); | |
| if (log.isDebugEnabled()) { | |
| log.debug("Loaded " + fsEventExecutors.size() + " event executors"); | |
| } |
Comment on lines
108
to
109
| consentPersist(consentPersistData, consentResource); | ||
| ConsentAuthorizeUtil.publishConsentApprovalStatus(consentPersistData); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 20
Suggested change
| consentPersist(consentPersistData, consentResource); | |
| ConsentAuthorizeUtil.publishConsentApprovalStatus(consentPersistData); | |
| consentPersist(consentPersistData, consentResource); | |
| log.info("Consent persisted successfully for consent ID: " + consentResource.getConsentID()); | |
| ConsentAuthorizeUtil.publishConsentApprovalStatus(consentPersistData); |
Comment on lines
188
to
189
| persistConsent(responseConsentResource, consentData); | ||
| ConsentAuthorizeUtil.publishConsentApprovalStatus(consentPersistData); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 21
Suggested change
| persistConsent(responseConsentResource, consentData); | |
| ConsentAuthorizeUtil.publishConsentApprovalStatus(consentPersistData); | |
| ExternalAPIConsentResourceResponseDTO responseConsentResource = responseDTO.getConsentResource(); | |
| persistConsent(responseConsentResource, consentData); | |
| log.info("Consent persisted successfully for consent ID: " + consentData.getConsentId()); | |
| ConsentAuthorizeUtil.publishConsentApprovalStatus(consentPersistData); |
Comment on lines
190
to
191
|
|
||
| } catch (FinancialServicesException e) { |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 22
Suggested change
| } catch (FinancialServicesException e) { | |
| } catch (FinancialServicesException e) { | |
| log.error("Failed to persist consent: " + e.getMessage()); |
Comment on lines
676
to
677
| public static void publishConsentApprovalStatus(ConsentPersistData consentPersistData) { | ||
| if (Boolean.parseBoolean((String) FinancialServicesConfigParser.getInstance().getConfiguration() |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 23
Suggested change
| public static void publishConsentApprovalStatus(ConsentPersistData consentPersistData) { | |
| if (Boolean.parseBoolean((String) FinancialServicesConfigParser.getInstance().getConfiguration() | |
| public static void publishConsentApprovalStatus(ConsentPersistData consentPersistData) { | |
| log.info("Publishing consent approval status for consent: " + consentPersistData.getConsentData().getConsentId()); | |
| if (Boolean.parseBoolean((String) FinancialServicesConfigParser.getInstance().getConfiguration() |
Comment on lines
702
to
703
| FSDataPublisherUtil.publishData("", "", consentAuthorizationData); | ||
| } else { |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 24
Suggested change
| FSDataPublisherUtil.publishData("", "", consentAuthorizationData); | |
| } else { | |
| FSDataPublisherUtil.publishData("", "", consentAuthorizationData); | |
| log.debug("Successfully published consent authorization data for consent: " + consentData.getConsentId()); | |
| } else { |
Comment on lines
+735
to
+739
|
|
||
| if (StringUtils.isBlank(requestUri)) { | ||
| log.error("Request URI not found."); | ||
| return null; | ||
| } |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 25
Suggested change
| if (StringUtils.isBlank(requestUri)) { | |
| log.error("Request URI not found."); | |
| return null; | |
| } | |
| if (StringUtils.isBlank(requestUri)) { | |
| log.error("Request URI not found."); | |
| return null; | |
| } | |
| if (log.isDebugEnabled()) { | |
| log.debug("Extracting request URI key from request URI"); | |
| } |
Comment on lines
+742
to
+744
| String requestUriKey = uriParts[uriParts.length - 1]; | ||
|
|
||
| return StringUtils.isBlank(requestUriKey) ? null : requestUriKey; |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 26
Suggested change
| String requestUriKey = uriParts[uriParts.length - 1]; | |
| return StringUtils.isBlank(requestUriKey) ? null : requestUriKey; | |
| String requestUriKey = uriParts[uriParts.length - 1]; | |
| if (StringUtils.isBlank(requestUriKey)) { | |
| log.warn("Request URI key is blank after extraction"); | |
| } | |
| return StringUtils.isBlank(requestUriKey) ? null : requestUriKey; |
Comment on lines
65
to
68
| @Override | ||
| public void processEvent(FSEvent fsEvent) { | ||
|
|
||
| Map<String, Object> eventData = fsEvent.getEventData(); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 27
Suggested change
| @Override | |
| public void processEvent(FSEvent fsEvent) { | |
| Map<String, Object> eventData = fsEvent.getEventData(); | |
| @Override | |
| public void processEvent(FSEvent fsEvent) { | |
| log.info("Processing consent lifecycle event"); | |
| Map<String, Object> eventData = fsEvent.getEventData(); |
Comment on lines
78
to
79
| log.debug("Publishing consent data for metrics."); | ||
|
|
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 28
Suggested change
| log.debug("Publishing consent data for metrics."); | |
| String consentId = (String) eventData.get(CONSENT_ID); | |
| log.info("Processing consent lifecycle event for consent ID: " + consentId); | |
| String primaryUserId; |
Comment on lines
+82
to
+84
| try { | ||
| primaryUserId = getPrimaryUserForConsent(detailedConsentResource, consentId); | ||
| } catch (ConsentManagementException e) { |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 29
Suggested change
| try { | |
| primaryUserId = getPrimaryUserForConsent(detailedConsentResource, consentId); | |
| } catch (ConsentManagementException e) { | |
| } catch (ConsentManagementException e) { | |
| log.error("Error while trying to retrieve consent data: " + e.getMessage()); | |
| return; |
Comment on lines
+87
to
+89
| } | ||
|
|
||
| if (StringUtils.isBlank(primaryUserId)) { |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 30
Suggested change
| } | |
| if (StringUtils.isBlank(primaryUserId)) { | |
| if (StringUtils.isBlank(primaryUserId)) { | |
| log.warn("Primary user ID is blank for consent ID: " + consentId); | |
| return; |
Comment on lines
+110
to
+112
| } | ||
| publishedEventIdentifierCache.put(eventIdentifier, Boolean.TRUE); | ||
| } |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 31
Suggested change
| } | |
| publishedEventIdentifierCache.put(eventIdentifier, Boolean.TRUE); | |
| } | |
| } | |
| } | |
| log.info("Publishing consent lifecycle data for consent ID: " + consentId); | |
| //publish consent lifecycle data |
Comment on lines
+131
to
+134
| } | ||
| } | ||
|
|
||
| private String getPrimaryUserForConsent(DetailedConsentResource detailedConsentResource, String consentId) |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 32
Suggested change
| } | |
| } | |
| private String getPrimaryUserForConsent(DetailedConsentResource detailedConsentResource, String consentId) | |
| private String getPrimaryUserForConsent(DetailedConsentResource detailedConsentResource, String consentId) | |
| throws ConsentManagementException { | |
| if (log.isDebugEnabled()) { | |
| log.debug("Retrieving primary user for consent ID: " + consentId); | |
| } | |
| String primaryUser = null; |
Comment on lines
34
to
38
| @Override | ||
| public FinancialServicesDataPublisher create() { | ||
|
|
||
| return (FinancialServicesDataPublisher) FSAnalyticsDataHolder.getInstance().getFinancialServicesDataPublisher(); | ||
| } |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 33
Suggested change
| @Override | |
| public FinancialServicesDataPublisher create() { | |
| return (FinancialServicesDataPublisher) FSAnalyticsDataHolder.getInstance().getFinancialServicesDataPublisher(); | |
| } | |
| @Override | |
| public FinancialServicesDataPublisher create() { | |
| log.debug("Creating new FinancialServicesDataPublisher instance"); | |
| return (FinancialServicesDataPublisher) FSAnalyticsDataHolder.getInstance().getFinancialServicesDataPublisher(); | |
| } |
Comment on lines
40
to
45
| @Override | ||
| public PooledObject<FinancialServicesDataPublisher> wrap(FinancialServicesDataPublisher | ||
| financialServicesDataPublisher) { | ||
|
|
||
| return new DefaultPooledObject<FinancialServicesDataPublisher>(financialServicesDataPublisher); | ||
| } |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 34
Suggested change
| @Override | |
| public PooledObject<FinancialServicesDataPublisher> wrap(FinancialServicesDataPublisher | |
| financialServicesDataPublisher) { | |
| return new DefaultPooledObject<FinancialServicesDataPublisher>(financialServicesDataPublisher); | |
| } | |
| @Override | |
| public PooledObject<FinancialServicesDataPublisher> wrap(FinancialServicesDataPublisher | |
| financialServicesDataPublisher) { | |
| log.debug("Wrapping FinancialServicesDataPublisher instance in pooled object"); | |
| return new DefaultPooledObject<FinancialServicesDataPublisher>(financialServicesDataPublisher); | |
| } |
Comment on lines
28
to
36
| */ | ||
| public class DataPublisherPool<FinancialServicesDataPublisher> extends | ||
| GenericObjectPool<FinancialServicesDataPublisher> { | ||
|
|
||
| public DataPublisherPool(PooledObjectFactory<FinancialServicesDataPublisher> factory, | ||
| GenericObjectPoolConfig<FinancialServicesDataPublisher> config) { | ||
|
|
||
| super(factory, config); | ||
| } |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 35
Suggested change
| */ | |
| public class DataPublisherPool<FinancialServicesDataPublisher> extends | |
| GenericObjectPool<FinancialServicesDataPublisher> { | |
| public DataPublisherPool(PooledObjectFactory<FinancialServicesDataPublisher> factory, | |
| GenericObjectPoolConfig<FinancialServicesDataPublisher> config) { | |
| super(factory, config); | |
| } | |
| public class DataPublisherPool<FinancialServicesDataPublisher> extends | |
| GenericObjectPool<FinancialServicesDataPublisher> { | |
| private static final Log log = LogFactory.getLog(DataPublisherPool.class); | |
| public DataPublisherPool(PooledObjectFactory<FinancialServicesDataPublisher> factory, | |
| GenericObjectPoolConfig<FinancialServicesDataPublisher> config) { | |
| super(factory, config); | |
| log.info("DataPublisherPool initialized with max total: " + config.getMaxTotal() + | |
| ", max idle: " + config.getMaxIdle()); | |
| } |
Comment on lines
40
to
45
| public EventQueue(int queueSize, int workerThreadCount) { | ||
|
|
||
| // Note : Using a fixed worker thread pool and a bounded queue to control the load on the server | ||
| publisherExecutorService = Executors.newFixedThreadPool(workerThreadCount); | ||
| eventQueue = new ArrayBlockingQueue<>(queueSize); | ||
| } |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 36
Suggested change
| public EventQueue(int queueSize, int workerThreadCount) { | |
| // Note : Using a fixed worker thread pool and a bounded queue to control the load on the server | |
| publisherExecutorService = Executors.newFixedThreadPool(workerThreadCount); | |
| eventQueue = new ArrayBlockingQueue<>(queueSize); | |
| } | |
| public EventQueue(int queueSize, int workerThreadCount) { | |
| log.info("Initializing EventQueue with queueSize: " + queueSize + " and workerThreadCount: " + workerThreadCount); | |
| // Note : Using a fixed worker thread pool and a bounded queue to control the load on the server | |
| publisherExecutorService = Executors.newFixedThreadPool(workerThreadCount); | |
| eventQueue = new ArrayBlockingQueue<>(queueSize); | |
| } |
Comment on lines
47
to
57
| public void put(FSAnalyticsEvent fsAnalyticsEvent) { | ||
|
|
||
| try { | ||
| if (eventQueue.offer(fsAnalyticsEvent)) { | ||
| publisherExecutorService.submit(new QueueWorker(eventQueue, publisherExecutorService)); | ||
| } else { | ||
| log.error("Event queue is full. Starting to drop OB analytics events."); | ||
| } | ||
| } catch (RejectedExecutionException e) { | ||
| log.warn("Task submission failed. Task queue might be full", e); | ||
| } |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 37
Suggested change
| public void put(FSAnalyticsEvent fsAnalyticsEvent) { | |
| try { | |
| if (eventQueue.offer(fsAnalyticsEvent)) { | |
| publisherExecutorService.submit(new QueueWorker(eventQueue, publisherExecutorService)); | |
| } else { | |
| log.error("Event queue is full. Starting to drop OB analytics events."); | |
| } | |
| } catch (RejectedExecutionException e) { | |
| log.warn("Task submission failed. Task queue might be full", e); | |
| } | |
| public void put(FSAnalyticsEvent fsAnalyticsEvent) { | |
| if (log.isDebugEnabled()) { | |
| log.debug("Attempting to add analytics event to queue"); | |
| } | |
| try { | |
| if (eventQueue.offer(fsAnalyticsEvent)) { | |
| publisherExecutorService.submit(new QueueWorker(eventQueue, publisherExecutorService)); | |
| } else { | |
| log.error("Event queue is full. Starting to drop OB analytics events."); | |
| } | |
| } catch (RejectedExecutionException e) { | |
| log.warn("Task submission failed. Task queue might be full", e); | |
| } | |
| } |
Member
Author
There was a problem hiding this comment.
added,without isDebugEnabled check
Comment on lines
59
to
63
|
|
||
| @Override | ||
| protected void finalize() throws Throwable { | ||
| publisherExecutorService.shutdown(); | ||
| super.finalize(); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 38
Suggested change
| @Override | |
| protected void finalize() throws Throwable { | |
| publisherExecutorService.shutdown(); | |
| super.finalize(); | |
| @Override | |
| protected void finalize() throws Throwable { | |
| log.info("Shutting down publisher executor service"); | |
| publisherExecutorService.shutdown(); | |
| super.finalize(); | |
| } |
Comment on lines
55
to
57
|
|
||
| public FinancialServicesThriftDataPublisher() { | ||
| this.init(); |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 39
Suggested change
| public FinancialServicesThriftDataPublisher() { | |
| this.init(); | |
| public FinancialServicesThriftDataPublisher() { | |
| log.info("Initializing FinancialServicesThriftDataPublisher"); | |
| this.init(); | |
| } |
Comment on lines
+59
to
+60
|
|
||
| @Override |
Contributor
There was a problem hiding this comment.
Log Improvement Suggestion No: 40
Suggested change
| @Override | |
| @Override | |
| public void publish(String streamName, String streamVersion, Map<String, Object> analyticsData) { | |
| if (log.isDebugEnabled()) { | |
| log.debug("Publishing data for stream: " + streamName + " with version: " + streamVersion); | |
| } |
Contributor
There was a problem hiding this comment.
AI Agent Log Improvement Checklist
- The log-related comments and suggestions in this review were generated by an AI tool to assist with identifying potential improvements. Purpose of reviewing the code for log improvements is to improve the troubleshooting capabilities of our products.
- Please make sure to manually review and validate all suggestions before applying any changes. Not every code suggestion would make sense or add value to our purpose. Therefore, you have the freedom to decide which of the suggestions are helpful.
✅ Before merging this pull request:
- Review all AI-generated comments for accuracy and relevance.
- Complete and verify the table below. We need your feedback to measure the accuracy of these suggestions and the value they add. If you are rejecting a certain code suggestion, please mention the reason briefly in the suggestion for us to capture it.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Pull Request Title
Issue link: required
Doc Issue: Optional, link issue from documentation repository
Applicable Labels: Spec, product, version, type (specify requested labels)
Development Checklist
Testing Checklist