Skip to content

Instantly share code, notes, and snippets.

View vmarcinko's full-sized avatar

Vjeran Marcinko vmarcinko

1 follower · 0 following
View GitHub Profile
@vmarcinko
vmarcinko / QueueConsumer.java
Last active June 3, 2019 15:34
public QueueConsumer(
QueueConsumerModule<?> queueConsumerModule,
RetryPolicy retryPolicy,
PlatformTransactionManager transactionManager,
int polledItemsLimit,
long pollingPeriodInSecs,
int partitionCount
) {
// ...
this.scheduledExecutorService = Executors.newScheduledThreadPool(partitionCount);
@vmarcinko
vmarcinko / QueueConsumerModule.java
Created June 3, 2019 15:26
public interface QueueConsumerModule<ID> {
List<ID> findItemIdsWhereQueueingNextAttemptTimeIsBefore(int partition, LocalDateTime time, int limit);
// ...
}
@vmarcinko
vmarcinko / SmsMessage.java
Last active June 3, 2019 15:25
@Entity
@Table(name = "sms_message", indexes = @Index(name = "idx_sms_msg_queue_polling_fields", columnList = "next_attempt_time,partition"))
public class SmsMessage {
// ...
@Column(name = "partition")
private int partition;
// ...
@vmarcinko
vmarcinko / QueueConsumerModule.java
Created June 3, 2019 15:13
public interface QueueConsumerModule<ID> {
List<ID> findItemIdsWhereQueueingNextAttemptTimeIsBefore(String fromAddress, LocalDateTime time, int limit);
// ...
}
@vmarcinko
vmarcinko / QueueConsumer.java
Created June 3, 2019 14:56
public QueueConsumer(
QueueConsumerModule<?> queueConsumerModule,
RetryPolicy retryPolicy,
PlatformTransactionManager transactionManager,
int polledItemsLimit,
long pollingPeriodInSecs,
int processingThreadCount
) {
// ...
this.processingExecutorService = Executors.newFixedThreadPool(processingThreadCount);
@vmarcinko
vmarcinko / QueueConsumer.java
Created May 31, 2019 08:33
@EventListener(OnGrantedEvent.class)
public void onGrantedEvent() {
logger.info("Granted leadership");
startProcessingTask();
}
@EventListener(OnRevokedEvent.class)
public void onRevokedEvent() {
logger.info("Revoked leadership");
@vmarcinko
vmarcinko / SmsQueueingApplication.java
Created May 31, 2019 08:29
public class SmsQueueingApplication {
// ...
@Bean
public DefaultLockRepository lockRepository(DataSource dataSource) {
return new DefaultLockRepository(dataSource);
}
@Bean
public JdbcLockRegistry lockRegistry(LockRepository lockRepository) {
@vmarcinko
vmarcinko / sms_queueing_log_4.txt
Created May 22, 2019 10:41
2019-05-10 07:45:12.160 ERROR 8415 --- [pool-1-thread-1] com.ag04.jpaqueue.QueueConsumer : Error while processing item by ID 316: Wrong time picked for send
java.lang.IllegalStateException: Wrong time picked for send
at com.ag04.smsqueueing.sender.SmsSenderImpl.send(SmsSenderImpl.java:23) ~[classes/:na]
at com.ag04.smsqueueing.SmsSendingQueueConsumerModule.processItem(SmsSendingQueueConsumerModule.java:50) ~[classes/:na]
at com.ag04.smsqueueing.SmsSendingQueueConsumerModule.processItem(SmsSendingQueueConsumerModule.java:16) ~[classes/:na]
at com.ag04.jpaqueue.QueueConsumer.processItem(QueueConsumer.java:123) [classes/:na]
at com.ag04.jpaqueue.QueueConsumer.lambda$processItemAndHandleErrorIfRaised$0(QueueConsumer.java:107) [classes/:na]
at com.ag04.jpaqueue.QueueConsumer$1.doInTransactionWithoutResult(QueueConsumer.java:117) ~[classes/:na]
at org.springframework.transaction.support.TransactionCallbackWithoutResult.doInTransaction(TransactionCallbackWithoutResult.java:36) ~[spring-tx-5.1.5.RELE
@vmarcinko
vmarcinko / sms_queueing_log_3.txt
Created May 22, 2019 10:39
2019-05-10 07:45:11.518 INFO 8415 --- [pool-1-thread-1] c.ag04.smsqueueing.sender.SmsSenderImpl : Sending SMS: SmsMessage[id=311, uid='e2a334af-4398-420a-b493-e5a05549fcc9', fromAddress='80444', toAddress='385913344829', text='Hello, this is text generated at 2019-05-10T07:40:36.360', sendingState=QueueingState[status=NOT_ATTEMPTED, nextAttemptTime=2019-05-10T07:40:36.360, attemptCount=0, lastAttemptTime=null, lastAttemptErrorMessage='null']]
2019-05-10 07:45:11.605 INFO 8415 --- [pool-1-thread-1] c.ag04.smsqueueing.sender.SmsSenderImpl : Sending SMS: SmsMessage[id=312, uid='dda97da8-af60-4eca-87d7-c406bf1c3161', fromAddress='80555', toAddress='385913344533', text='Hello, this is text generated at 2019-05-10T07:40:36.363', sendingState=QueueingState[status=NOT_ATTEMPTED, nextAttemptTime=2019-05-10T07:40:36.363, attemptCount=0, lastAttemptTime=null, lastAttemptErrorMessage='null']]
2019-05-10 07:45:11.699 INFO 8415 --- [pool-1-thread-1] c.ag04.smsqueueing.sender.SmsSenderImpl : Sending SMS: SmsMessage[id
@vmarcinko
vmarcinko / sms_queueing_log_2.txt
Created May 22, 2019 08:10
2019-05-10 07:45:11.512 INFO 8415 --- [pool-1-thread-1] com.ag04.jpaqueue.QueueConsumer : Fetched 100 pending queued items