Distributed Scheduling with Spring Boot: the challenges and pitfalls of implementing a background job

(🎥 RECORDING: https://www.youtube.com/watch?v=ghpljMg8Ecc)

Sooner or later a developer will implement his/her first background job using Java and Spring Boot, and what usually is a simple task for the majority of systems might become a nightmare in scenarios that need to deal with high performance, parallelism, distributed systems and a large volume of data. Scenarios like those hide several issues which many developers are not used to, such as large volumes of data, network failures, data inconsistency, out-of-memory errors and even taking the whole system down.

Although it seems controversial, dealing with many of these problems does not require hype technologies or services, but solid distributed systems fundamentals. This talk will present how an experienced developer implements a background job with Java and Spring Boot taking into consideration the main challenges and pitfalls it brings along, and how he/she designs a solution for high-performance, resilience and horizontal scalability at the same time he/she takes advantage of many modules of Spring Boot, Hibernate and the relational database.

If you still believe that a background job is a simple task, so this talk is for you!

1. DISTRIBUTED SCHEDULING | Spring Boot the challenges and pitfalls of implementing a background job Rafael Ponte | @rponte
2. https://twitter.com/rponte/status/1492168535421992960
3. But implementing a background job…
5. https://deniseyu.io/art/
6. Concurrent and Distributed Computing
8. Where is my Credit Card?
13. db server
14. db server
15. db server
16. db server
17. db server
18. db server
19. db server
20. db server
21. db server
25. How to schedule tasks with Spring Boot?
26. @Component public class OneJob { @Scheduled(fixedDelay = 60_000) public void runQuiteOften() { // lógica do job vai aqui } }
27. @Component public class OneJob { @Scheduled(fixedDelay = 60_000) public void runQuiteOften() { // lógica do job vai aqui } }
28. @Component public class OneJob { @Scheduled(fixedDelay = 60_000) public void runQuiteOften() { // job logic goes here } }
29. @Component public class OneJob { @Scheduled(fixedDelay = 60_000) public void runQuiteOften() { // job logic goes here } }
30. @Component public class OneJob { public void runQuiteOften() { // job logic goes here } }
31. @Component public class OneJob { @Scheduled public void runQuiteOften() { // job logic goes here } }
32. @Component public class OneJob { @Scheduled(fixedDelay = 60_000) public void runQuiteOften() { // job logic goes here } }
33. @Component public class OneJob { @Scheduled(fixedDelay = 60_000) public void runQuiteOften() { // job logic goes here } }
34. Coming back…
35. db server
36. server db
37. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
38. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
39. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
40. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
41. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
42. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
43. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
44. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
45. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
46. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
47. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
48. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
49. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
50. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
51. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } } Naive Solution
52. @rponte Rafael Ponte Brazilian Software Engineer at aka Prince of the Ocean aka Maharaja of the Legacies 19+ years of experience working with Java and distributed systems Loves working with "boring techs"
53. DISTRIBUTED SCHEDULING | Spring Boot the challenges and pitfalls of implementing a background job Rafael Ponte | @rponte
54. Naive Solution
55. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
56. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
57. select p.* from proposal p where p.status = 'ELIGIBLE' order by p.created_at asc
58. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
59. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
60. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
61. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
62. What if…
63. …the application crashes?
64. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
65. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
66. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
67. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
68. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } Inserts a new card Updates a proposal with its card
69. -- Inserts a new credit card INSERT INTO card (id, cardNumber) VALUES (42, ’5260-9991-9040-0001’); -- Updates the proposal with its credit card UPDATE proposal SET card_id = 42, status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 7031;
70. -- Inserts a new credit card INSERT INTO card (id, cardNumber) VALUES (42, ’5260-9991-9040-0001’); -- Updates the proposal with its credit card UPDATE proposal SET card_id = 42, status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 7031;
71. -- Inserts a new credit card INSERT INTO card (id, cardNumber) VALUES (42, ’5260-9991-9040-0001’); -- Updates the proposal with its credit card UPDATE proposal SET card_id = 42, status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 7031;
72. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
73. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); // begins and commits a transaction proposal.attachTo(newCard); proposalRepository.save(proposal); // begins and commits a transaction }); }
74. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); // begins and commits a transaction proposal.attachTo(newCard); proposalRepository.save(proposal); // begins and commits a transaction }); } it’s similar to auto_commit = true😱
75. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
76. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
77. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
78. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
79. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
80. ID STATUS CARD_ID 7028 NOT_ELIGIBLE 39 7029 NOT_ELIGIBLE 40 7030 ELIGIBLE_WITH_ATTACHED_CARD 41 7031 ELIGIBLE <null> 7032 ELIGIBLE_WITH_ATTACHED_CARD 96 ... ... ... ID CARD_NUMBER 40 5260-9991-9040-0040 41 5260-9991-9040-0041 42 5260-9991-9040-00042 Proposal Card Database
81. ID STATUS CARD_ID 7028 NOT_ELIGIBLE 39 7029 NOT_ELIGIBLE 40 7030 ELIGIBLE_WITH_ATTACHED_CARD 41 7031 ELIGIBLE <null> 7032 ELIGIBLE_WITH_ATTACHED_CARD 96 ... ... ... ID CARD_NUMBER 40 5260-9991-9040-0040 41 5260-9991-9040-0041 42 5260-9991-9040-00042 Proposal Card Database
82. ID STATUS CARD_ID 7028 NOT_ELIGIBLE 39 7029 NOT_ELIGIBLE 40 7030 ELIGIBLE_WITH_ATTACHED_CARD 41 7031 ELIGIBLE <null> 7032 ELIGIBLE_WITH_ATTACHED_CARD 96 ... ... ... ID CARD_NUMBER 40 5260-9991-9040-0040 41 5260-9991-9040-0041 42 5260-9991-9040-00042 Proposal Card Database
83. ID STATUS CARD_ID 7028 NOT_ELIGIBLE 39 7029 NOT_ELIGIBLE 40 7030 ELIGIBLE_WITH_ATTACHED_CARD 41 7031 ELIGIBLE <null> 7032 ELIGIBLE_WITH_ATTACHED_CARD 96 ... ... ... ID CARD_NUMBER 40 5260-9991-9040-0040 41 5260-9991-9040-0041 42 5260-9991-9040-00042 Proposal Card Database
84. Maybe…
85. https://twitter.com/rponte/status/1492168535421992960
86. Runs every 1 minute ID COL1 COL2 1 2 3 4 ID COL1 COL2 1 2 3 4 Data corruption error
87. Runs every 1 minute Runs every 5 minutes ID COL1 COL2 1 2 3 4 ID COL1 COL2 1 2 3 4 ID COL1 COL2 1 2 3 4 ID COL1 COL2 1 2 3 4 Data corruption error Fix corrupted data
88. Runs every 1 minute Runs every 5 minute ID COL1 COL2 1 2 3 4 ID COL1 COL2 1 2 3 4 ID COL1 COL2 1 2 3 4 ID COL1 COL2 1 2 3 4 Data corruption error Fix corrupted data No!
89. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
90. @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
92. @Transactional @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
93. @Transactional @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
94. @Transactional @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); }
95. @Transactional @Scheduled(…) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderBy…; proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(…); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } You got Recoverability!
96. What if…
97. …data volume grows?
98. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
99. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
100. select p.* from proposal p where p.status = 'ELIGIBLE' order by p.created_at asc
103. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
104. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findAllByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
105. @Repository public interface ProposalRepository extends JpaRepository<Proposal, UUID> { public List<Proposal> findAllByStatusOrderByCreatedAtAsc(ProposalStatus status); }
106. @Repository public interface ProposalRepository extends JpaRepository<Proposal, UUID> { public List<Proposal> findAllByStatusOrderByCreatedAtAsc(ProposalStatus status); }
107. @Repository public interface ProposalRepository extends JpaRepository<Proposal, UUID> { public List<Proposal> findAllByStatusOrderByCreatedAtAsc(ProposalStatus status); }
108. @Repository public interface ProposalRepository extends JpaRepository<Proposal, UUID> { public List<Proposal> findAllByStatusOrderByCreatedAtAsc(ProposalStatus status); }
109. select p.* from proposal p where p.status = 'ELIGIBLE' order by p.created_at asc
110. List<Proposal> findAllByStatusOrderByCreatedAtAsc(...);
111. List<Proposal> findAllByStatusOrderByCreatedAtAsc(...);
112. List<Proposal> findTop50ByStatusOrderByCreatedAtAsc(...);
113. select p.* from proposal p where p.status = 'ELIGIBLE' order by p.created_at asc
114. select p.* from proposal p where p.status = 'ELIGIBLE' order by p.created_at asc limit 50
115. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
116. But…
117. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } }
118. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } } this loop will end very EARLY!
119. @Component public class AttachCardsToProposalsJob { private CardsClient cardsClient; private CardRepository cardRepository; private ProposalRepository proposalRepository; @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } } It will only run on the NEXT execution!
120. while(true)
121. @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); proposals.forEach(proposal -> { // process each proposal }); }
122. @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { // process each proposal }); } }
123. @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { // process each proposal }); } }
124. @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { // process each proposal }); } }
125. But after running…
127. What?
128. @Component public class AttachCardsToProposalsJob { // dependencies and constants @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } } }
129. @Component public class AttachCardsToProposalsJob { // dependencies and constants @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } } } - fetching only 50 objects at a time
130. @Component public class AttachCardsToProposalsJob { // dependencies and constants @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } } } - creating a new reference - fetching only 50 objects at a time
131. @Component public class AttachCardsToProposalsJob { // dependencies and constants @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } } } - exiting the loop at the right time - creating a new reference - fetching only 50 objects at a time
132. Why?
133. @Component public class AttachCardsToProposalsJob { // dependências e constantes @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } } }
134. @Repository public interface ProposalRepository extends JpaRepository<Proposal, UUID> { public List<Proposal> findAllByStatusOrderByCreatedAtAsc(ProposalStatus status); }
135. Revisiting JPA and Hibernate
136. Job
137. Repository Job
138. Repository EM Job
139. Repository EM Job
140. Repository EM Job
141. Repository EM Job
142. Repository EM Job
143. Repository EM Job
144. Repository EM Job
145. That’s what we expected…
146. Long-running transaction
147. Repository EM Job Long-running transaction
148. @Transactional Repository EM Job Long-running transaction
149. @Transactional Repository EM Job Long-running transaction
150. @Transactional Repository EM Job Long-running transaction
151. @Transactional Repository EM Job (First-Level Cache) Long-running transaction
152. @Transactional Repository EM Job (First-Level Cache) Long-running transaction
153. The idea here is to understand…
154. @Transactional Repository EM Job (First-Level Cache) Long-running transaction
155. @Transactional EM (First-Level Cache) Long-running transaction
156. @Transactional EM (First-Level Cache) Long-running transaction
157. EM (First-Level Cache) @Transactional Long-running transaction
158. EM (First-Level Cache) @Transactional Long-running transaction
159. Repository EM Job (First-Level Cache) @Transactional Long-running transaction
160. Repository EM Job (First-Level Cache) Long-running transaction
161. Ahhh!!
162. @Component public class AttachCardsToProposalsJob { // dependencies and constants @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } } }
163. @Component public class AttachCardsToProposalsJob { // dependencies and constants @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } } } begins and ends a transaction
164. But…
165. @Component public class AttachCardsToProposalsJob { // dependencies and constants @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); } } }
166. @Component public class AttachCardsToProposalsJob { // dependencies and constants @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); // begins and commits a transaction proposal.attachTo(newCard); proposalRepository.save(proposal); // begins and commits a transaction }); } } } We gave up on data consistency! 🤡
167. @Component public class AttachCardsToProposalsJob { // dependencies and constants @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(ELIGIBLE); if (proposals.isEmpty()) { break; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); // begins and commits a transaction proposal.attachTo(newCard); proposalRepository.save(proposal); // begins and commits a transaction }); } } } We give up on data consistency! Nooo!
168. Short-running transaction
169. Job Transaction Job
170. Job Transaction Job start of job end of job
171. Job Transaction Job Tx1
172. start of transaction end of transaction Job Job Transaction Tx1
173. Job Transaction Job Long-running transaction Tx1
174. We need to minimize it…
175. Job Transaction Job
176. Job Transaction Job Tx1
177. Job Transaction Job start of transaction end of transaction Tx1
178. Tx1 Job Transaction Job Short-running transaction
179. Job Transaction Job Tx1
180. Job Transaction Job Tx1 Tx2
181. Job Transaction Job Tx1 Tx2 Tx3
182. Job Transaction Job Tx1 Tx2 Tx3 Tx4
183. Job Transaction Job Tx1 Tx2 Tx3 Tx4 Tx5 Tx6 TxN
184. Job Transaction Job Short-running transaction Tx1 Tx2 Tx3 Tx4 Tx5 Tx6 TxN
185. Repository EM Job (First-Level Cache) Short-running transaction
186. Repository EM Job @Transactional Short-running transaction (First-Level Cache)
187. Repository EM Job @Transactional Short-running transaction (First-Level Cache)
188. Repository EM Job @Transactional Short-running transaction (First-Level Cache)
189. Repository EM Job @Transactional Short-running transaction (First-Level Cache)
190. Transaction Management PROGRAMMATIC
191. @Component public class AttachCardsToProposalsJob { @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { // executes the logic within a transactional scope } } }
192. @Component public class AttachCardsToProposalsJob { @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { // executes the logic within a transactional scope } } }
193. @Component public class AttachCardsToProposalsJob { @Transactional @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { @Transactional.execute(transaction -> { // executes the logic within a transactional scope });; } } }
194. @Component public class AttachCardsToProposalsJob { @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { @Transactional.execute(transaction -> { // executes the logic within a transactional scope }); } } }
195. @Component public class AttachCardsToProposalsJob { @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { @Transactional.execute(transaction -> { // executes the logic within a transactional scope }); } } }
196. @Component public class AttachCardsToProposalsJob { @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { @Transactional.execute(transaction -> { // executes the logic within a transactional scope }); } } }
197. @Component public class AttachCardsToProposalsJob { @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { @Transactional.execute(transaction -> { // executes the logic within a transactional scope }); } } }
198. @Component public class AttachCardsToProposalsJob { @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { @Transactional.execute(transaction -> { // executes the logic within a transactional scope }); } } }
199. @Component public class AttachCardsToProposalsJob { @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { @Transactional.execute(transaction -> { // executes the logic within a transactional scope }); } } } It does NOT execute! 😵💫
200. @Component public class AttachCardsToProposalsJob { @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { @Transactional.execute(transaction -> { // executes the logic within a transactional scope }); } } }
201. @Component public class AttachCardsToProposalsJob { @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { transactionManager.execute(transaction -> { // executes the logic within a transactional scope }); } } }
202. @Component public class AttachCardsToProposalsJob { private TransactionTemplate transactionManager; @Scheduled(fixedDelay = 60_000) public void execute() { while (true) { transactionManager.execute(transaction -> { // executes the logic within a transactional scope }); } } }
203. After refactoring…
204. @Autowired private TransactionTemplate transactionManager; @Scheduled(fixedDelay = 60_000) public void execute() { boolean pending = true; while (pending) { pending = transactionManager.execute(transactionStatus -> { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { return false; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); return true; }); } }
206. start of job end of job
207. start of job end of job You got batch processing!
208. What if…
209. …we run the application in a cluster?
213. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE
214. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE
215. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE
216. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE
217. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE
218. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE 1004 ELIGIBLE
219. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE 1004 ELIGIBLE
220. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE_WITH_ATTACHED_CARD 1004 ELIGIBLE
221. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE_WITH_ATTACHED_CARD 1004 ELIGIBLE
222. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE_WITH_ATTACHED_CARD 1004 ELIGIBLE_WITH_ATTACHED_CARD
223. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE_WITH_ATTACHED_CARD 1004 ELIGIBLE_WITH_ATTACHED_CARD
225. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE
226. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE
227. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE
228. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE 2x
229. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE 1004 ELIGIBLE 2x
230. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE 1004 ELIGIBLE 2x 2x
231. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE_WITH_ATTACHED_CARD 1004 ELIGIBLE 2x 2x
232. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE_WITH_ATTACHED_CARD 1004 ELIGIBLE 2x 2x 2x
233. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE_WITH_ATTACHED_CARD 1004 ELIGIBLE_WITH_ATTACHED_CARD 2x 2x 2x
234. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE_WITH_ATTACHED_CARD 1004 ELIGIBLE_WITH_ATTACHED_CARD 2x 2x 2x 2x
235. Oops!
236. Synchronized?
237. @Component public class AttachCardsToProposalsJob { @Scheduled(...) public synchronized void execute() { // executes the logic here } }
242. @Component public class AttachCardsToProposalsJob { @Scheduled(...) public synchronized void execute() { // executes the logic here } } No!
243. Distributed Lock?
244. @Scheduled(...) public void execute() { if (isLeader()) { // executes the logic here } }
245. @Scheduled(...) public void execute() { if (isLeader()) { // executes the logic here } }
257. No!
258. How about…
261. @Repository public interface ProposalRepository extends JpaRepository<Proposal, UUID> { public List<Proposal> findTop50ByStatusOrderByCreatedAtAsc(ProposalStatus status); }
262. @Repository public interface ProposalRepository extends JpaRepository<Proposal, UUID> { @Lock(LockModeType.PESSIMISTIC_WRITE) public List<Proposal> findTop50ByStatusOrderByCreatedAtAsc(ProposalStatus status); }
263. select p.* from proposal p where p.status = 'ELIGIBLE' order by p.created_at asc limit 50
264. select p.* from proposal p where p.status = 'ELIGIBLE' order by p.created_at asc limit 50 for update
265. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE ... ...
266. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE ... ...
267. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE ... ...
268. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE ... ...
269. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE 1003 ELIGIBLE ... ...
270. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE 1003 ELIGIBLE ... ...
271. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE 1003 ELIGIBLE ... ...
272. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE 1003 ELIGIBLE ... ...
273. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE 1003 ELIGIBLE ... ...
274. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE 1003 ELIGIBLE ... ...
275. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
276. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
277. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
278. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
279. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE_WITH_ATTACHED_CARD ... ...
280. Yess!
281. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE_WITH_ATTACHED_CARD ... ... You got high-availability!
282. What if…
283. …we need more throughput?
291. To distribute the load…
292. 9630 9631 QUEUE 9629
293. 9630 9631 producer 9629 produces: 1000 items/min QUEUE
294. 9630 9631 producer consumer 9629 consumes: 600 items/min produces: 1000 items/min QUEUE
295. 9630 9631 producer consumer 9629 consumer consumes: 400 items/min consumes: 600 items/min produces: 1000 items/min QUEUE
296. consumes: 400 items/min consumes: 600 items/min produces: 1000 items/min 9630 9631 producer consumer 9629 consumer QUEUE
298. No!
299. How about…
301. @Repository public interface ProposalRepository extends JpaRepository<Proposal, UUID> { @QueryHints({ @QueryHint( name = "javax.persistence.lock.timeout", value = LockOptions.SKIP_LOCKED) // org.hibernate.LockOptions }) @Lock(LockModeType.PESSIMISTIC_WRITE) public List<Proposal> findTop50ByStatusOrderByCreatedAtAsc(ProposalStatus status); }
302. @Repository public interface ProposalRepository extends JpaRepository<Proposal, UUID> { @QueryHints({ @QueryHint( name = "javax.persistence.lock.timeout", value = LockOptions.SKIP_LOCKED) // org.hibernate.LockOptions }) @Lock(LockModeType.PESSIMISTIC_WRITE) public List<Proposal> findTop50ByStatusOrderByCreatedAtAsc(ProposalStatus status); }
303. @Repository public interface ProposalRepository extends JpaRepository<Proposal, UUID> { @QueryHints({ @QueryHint( name = "javax.persistence.lock.timeout", value = LockOptions.SKIP_LOCKED) // org.hibernate.LockOptions }) @Lock(LockModeType.PESSIMISTIC_WRITE) public List<Proposal> findTop50ByStatusOrderByCreatedAtAsc(ProposalStatus status); }
304. select p.* from proposal p where p.status = 'ELIGIBLE' order by p.created_at asc limit 50 for update
305. select p.* from proposal p where p.status = 'ELIGIBLE' order by p.created_at asc limit 50 for update skip locked
306. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE ... ...
307. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE ... ...
308. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE ... ...
309. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE ... ...
310. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE ... ...
311. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE ... ...
312. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE ... ...
313. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
314. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
315. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
316. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
317. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
318. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
319. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
320. ID STATUS 1001 ELIGIBLE_WITH_ATTACHED_CARD 1002 ELIGIBLE_WITH_ATTACHED_CARD 1003 ELIGIBLE ... ...
321. In other words…
322. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE
323. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE
324. Yess!
325. ID STATUS 1001 ELIGIBLE 1002 ELIGIBLE 1003 ELIGIBLE 1004 ELIGIBLE You got scalability!
326. What if…
327. …we need more performance?
328. @Autowired private TransactionTemplate transactionManager; @Scheduled(fixedDelay = 60_000) public void execute() { boolean pending = true; while (pending) { pending = transactionManager.execute(transactionStatus -> { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { return false; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); return true; }); } }
329. @Autowired private TransactionTemplate transactionManager; @Scheduled(fixedDelay = 60_000) public void execute() { boolean pending = true; while (pending) { pending = transactionManager.execute(transactionStatus -> { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { return false; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); return true; }); } }
330. @Autowired private TransactionTemplate transactionManager; @Scheduled(fixedDelay = 60_000) public void execute() { boolean pending = true; while (pending) { pending = transactionManager.execute(transactionStatus -> { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { return false; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); return true; }); } }
331. @Autowired private TransactionTemplate transactionManager; @Scheduled(fixedDelay = 60_000) public void execute() { boolean pending = true; while (pending) { pending = transactionManager.execute(transactionStatus -> { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { return false; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); return true; }); } }
332. @Autowired private TransactionTemplate transactionManager; @Scheduled(fixedDelay = 60_000) public void execute() { boolean pending = true; while (pending) { pending = transactionManager.execute(transactionStatus -> { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { return false; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); return true; }); } }
333. @Autowired private TransactionTemplate transactionManager; @Scheduled(fixedDelay = 60_000) public void execute() { boolean pending = true; while (pending) { pending = transactionManager.execute(transactionStatus -> { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { return false; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); return true; }); } } Inserts a new card Updates a proposal with its card
334. For each iteration in the loop…
335. 2 roundtrips INSERT INTO card(cardNumber) VALUES ('5260-9991-9040-0001'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2021;
336. 2 roundtrips INSERT INTO card(cardNumber) VALUES ('5260-9991-9040-0001'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2021;
337. 2 roundtrips INSERT INTO card(cardNumber) VALUES ('5260-9991-9040-0001'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2021;
338. INSERT INTO card(cardNumber) VALUES ('5260-9991-9040-0001'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2021; INSERT INTO card(cardNumber) VALUES (' 5513-8141-9261-0002'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2022; 4 roundtrips
339. INSERT INTO card(cardNumber) VALUES ('5260-9991-9040-0001'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2021; INSERT INTO card(cardNumber) VALUES (' 5513-8141-9261-0002'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2022; INSERT INTO card(cardNumber) VALUES (' 5599-4154-6354-0003'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2023; 6 roundtrips
340. 100 roundtrips INSERT INTO card(cardNumber) VALUES ('5260-9991-9040-0001'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2021; INSERT INTO card(cardNumber) VALUES (' 5513-8141-9261-0002'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2022; INSERT INTO card(cardNumber) VALUES (' 5599-4154-6354-0003'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2023; INSERT INTO card(cardNumber) VALUES (' 5419-8753-1126-0004'); UPDATE proposal SET status = 'ELIGIBLE_WITH_ATTACHED_CARD' WHERE id = 2024; INSERT INTO ... UPDATE proposal SET ...
341. Latency total=120ms server database 40ms 80ms
342. Hibernate batch size
343. spring: jpa: properties: hibernate: jdbc: batch_size: 50 order_inserts: true order_updates: true application.yml
344. spring: jpa: properties: hibernate: jdbc: batch_size: 50 order_inserts: true order_updates: true application.yml
345. For each batch…
346. 2 roundtrips INSERT INTO card(cardNumber) VALUES (?) Params: [ ('5513-8141-9261-0001'), ('5513-8141-9261-0002'), (‘5513-8141-9261-0003'), (‘5513-8141-9261-0004'), ... ] UPDATE proposal SET status = ? WHERE id = ?; Params: [ (‘ELIGIBLE_WITH_ATTACHED_CARD’, 2021), ('ELIGIBLE_WITH_ATTACHED_CARD', 2022), ('ELIGIBLE_WITH_ATTACHED_CARD', 2023), ('ELIGIBLE_WITH_ATTACHED_CARD', 2024), ... ]
347. 2 roundtrips INSERT INTO card(cardNumber) VALUES (?) Params: [ ('5513-8141-9261-0001'), ('5513-8141-9261-0002'), (‘5513-8141-9261-0003'), (‘5513-8141-9261-0004'), ... ] UPDATE proposal SET status = ? WHERE id = ?; Params: [ (‘ELIGIBLE_WITH_ATTACHED_CARD’, 2021), ('ELIGIBLE_WITH_ATTACHED_CARD', 2022), ('ELIGIBLE_WITH_ATTACHED_CARD', 2023), ('ELIGIBLE_WITH_ATTACHED_CARD', 2024), ... ]
348. INSERT INTO card(cardNumber) VALUES (?) Params: [ ('5513-8141-9261-0001'), ('5513-8141-9261-0002'), (‘5513-8141-9261-0003'), (‘5513-8141-9261-0004'), ... ] 2 roundtrips UPDATE proposal SET status = ? WHERE id = ?; Params: [ (‘ELIGIBLE_WITH_ATTACHED_CARD’, 2021), ('ELIGIBLE_WITH_ATTACHED_CARD', 2022), ('ELIGIBLE_WITH_ATTACHED_CARD', 2023), ('ELIGIBLE_WITH_ATTACHED_CARD', 2024), ... ]
349. INSERT INTO card(cardNumber) VALUES (?) Params: [ ('5513-8141-9261-0001'), ('5513-8141-9261-0002'), (‘5513-8141-9261-0003'), (‘5513-8141-9261-0004'), ... ] 2 roundtrips UPDATE proposal SET status = ? WHERE id = ?; Params: [ (‘ELIGIBLE_WITH_ATTACHED_CARD’, 2021), ('ELIGIBLE_WITH_ATTACHED_CARD', 2022), ('ELIGIBLE_WITH_ATTACHED_CARD', 2023), ('ELIGIBLE_WITH_ATTACHED_CARD', 2024), ... ] You got optimal batch processing!
350. What if…
351. …something goes wrong with the network?
352. @Scheduled(...) public void execute() { boolean pending = true; while (pending) { pending = transactionManager.execute(txStatus -> { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { return false; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); return true; }); } }
353. @Scheduled(...) public void execute() { boolean pending = true; while (pending) { pending = transactionManager.execute(txStatus -> { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { return false; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); return true; }); } }
354. @Scheduled(...) public void execute() { boolean pending = true; while (pending) { pending = transactionManager.execute(txStatus -> { List<Proposal> proposals = proposalRepository.findTop50ByStatusOrderByCreatedAtAsc(…); if (proposals.isEmpty()) { return false; } proposals.forEach(proposal -> { CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); Card newCard = cardData.toModel(); cardRepository.save(newCard); proposal.attachTo(newCard); proposalRepository.save(proposal); }); return true; }); } }
355. https://deniseyu.io/art/
356. CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId());
357. CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); waits in fi nitely?
358. CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); waits in fi nitely? timeout
359. CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); transient errors? waits in fi nitely? timeout
360. CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); transient errors? retry+backo f waits in fi nitely? timeout
361. CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); can I send the request again? transient errors? retry+backo f waits in fi nitely? timeout
362. CardDataResponse cardData = cardsClient.findCardByProposalId(proposal.getId()); can I send the request again? idempotency transient errors? retry+backo f waits in fi nitely? timeout
363. WHAT WHAT WHAT WHAT WHAT WHAT WHAT WHAT WHAT IF… WHAT IF… WHAT IF… WHAT WHAT IF… HAT IF… WHAT IF… WHAT WHAT IF… WHAT IF… WHAT IF… WHAT IF… AT IF… WHAT IF… F…WHAT
364. Concluding
365. A background job can be hard… 😬
366. Recap…
367. Now, we have a background job that…
369. …can recover from crashes
370. …can handle huge volumes of data …can recover from crashes
371. …remains available in the face of failures …can handle huge volumes of data …can recover from crashes
372. …designed and built to be scalable …remains available in the face of failures …can handle huge volumes of data …can recover from crashes
373. …designed and built to be scalable …remains available in the face of failures …can handle huge volumes of data …optimized for high-throughput …can recover from crashes
375. But it also works on all other platforms
376. @rponte Rafael Ponte Thank you!!