Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Reactive Transactions Masterclass

Reactive Transactions Masterclass

Joined talk with Mark Paluch (http://twitter.com/mp911de, https://speakerdeck.com/mp911de) at Devoxx 2019.

How are reactive transactions supposed to work in a non-blocking, reactive application? Spring draws with its reactive transaction manager a new, strong primitive in the picture of reactive systems.

We will deeply dive into Reactive Relational Database Connectivity, the reactive specification for SQL database access and into Neo4j 4 that comes with a reactive database client. We will walk through the access of strictly transactional data sources while embracing reactive and non-blocking properties.

This highly technical Deep Dive session will visit reactive patterns for potentially highly concurrent applications that are no longer opinionated about threading.

Come to this session and learn how to set up and use transactions in a reactive application. We will present R2DBC and Neo4j examples and are open for questions, comments, and discussion.

Michael Simons

November 05, 2019
Tweet

More Decks by Michael Simons

Other Decks in Programming

Transcript

  1. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Mark Paluch • Spring Data

    Project Lead • Lettuce Redis Driver Project Lead • R2DBC Contributor • All things Reactive Data & Open Source github.com/mp911de • @mp911de • paluch.biz
  2. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Michael Simons • Neo4j since

    July 2018 • Java Champion and Oracle Groundbreaker • Co-Founder and current lead of Java User Group EuregJUG • Author
  3. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 More books: Life behind bars

    • I run… • I bike… • …and take pictures Revenue will go to Médecins Sans Frontières msf.org this year https://leanpub.com/lifebehindbars
  4. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Transaction • Atomically apply changes

    • Isolation • Consistency • Durable • Require a transactional resource
  5. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Database Example • Open connection

    • Begin transaction • Apply work • Reuse same connection across operations • Commit/rollback transaction
  6. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Transaction Management • Ensures transactional

    behaviour • Apply transaction definition (propagation, attributes) • Implementations • JTA • Spring • Driver-specific
  7. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Transaction Walkthrough • Allocate connection

    • Bind connection to transaction (Thread*) • Prepare connection for transaction (isolation level, …) • Issue queries (Reuse bound connection) • Exit transactional scope • Cleanup, commit or rollback • Run synchronizations • Unbind resources
  8. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Bound connection • Transactional state

    associated with transport connection • Works since the ’70s like that • Alternatively: Session • Newer implementations: Portable session • MongoDB
  9. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Imperative Transactions • Stay on

    a single thread* • ThreadLocal storage • Using another thread breaks transactions * ManagedExecutorService, @Async
  10. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Time dependency • Begin Tx

    before work • Work after begin Tx • Work before cleanup Tx • Cleanup Tx after work
  11. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Distributed Transactions • XA specification

    • Released in the 90’s • Transactions across distributed services • Two phase commit • Turned out that it sucks
  12. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Still Distributed Transactions • Redis

    + JDBC Transaction • Kind-of works • Leading transaction manager • Additional transactional resources may join • Independent commit/rollback • Still sucks
  13. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Database • Open connection •

    Begin transaction • Apply work • Reuse same connection across operations • Commit/rollback transaction
  14. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Transaction Walkthrough • Allocate connection

    • Bind connection to transaction (Subscription) • Prepare connection for transaction (isolation level, …) • Issue queries (Reuse bound connection) • Exit transactional scope • Cleanup, commit or rollback • Run synchronizations • Unbind resources
  15. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Time dependency • Begin before

    work • Work after begin • Work before cleanup • Cleanup after work
  16. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Imperative vs. Reactive • Single-threaded

    processing required • ThreadLocal transaction status • Data stays within transactional bounds • Synchronized begin and cleanup • Multi-threaded processing possible • Subscription-bound transaction status • Data escapes transactional bounds • Synchronized begin and cleanup
  17. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Cancellation • .first(), .take(n) •

    Send cancellation signal upstream • „Thank you I don’t need additional data“ • „Connection dead. Cancelling“
  18. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Cancellation effects • Cancellation considered

    success • Cancelled subscription does not await completion • Commit may be not yet visible to subsequent operations
  19. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 // Define a flow that

    creates 1000 nodes in a database Flux<Integer> createNodes = Flux.using( driver::rxSession, session -> session.run( "UNWIND range (1,1000) AS i CREATE (s:SomeNode {pos: i}) return s").records(), RxSession::close ).map(r -> r.get("s").get("pos").asInt()); „Cancelled subscription does not await completion“ What does that even mean?
  20. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 // Just take some of

    those createNodes.take(5); „Cancelled subscription does not await completion“ What does that even mean?
  21. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 // Or more explicit createNodes

    .as(StepVerifier::create) .expectNext(1, 2, 3, 4, 5) .thenCancel() .verify(); „Cancelled subscription does not await completion“ What does that even mean?
  22. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 // Can we verify it?

    Flux.using( driver::rxSession, session -> session.run("MATCH (s:SomeNode) RETURN count(s) as cnt").records(), RxSession::close ) .map(r -> r.get("cnt").asLong()) .single().as(StepVerifier::create).expectNext(1000L).verifyComplete(); „Cancelled subscription does not await completion“ What does that even mean?
  23. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Cancellation • Happens when the

    browser goes away • Probably not a good idea to return transactional flows from web endpoints (The same has been said for @Transactional on web methods for a long time!) • Again: .first(), .take(n) • Also limitRequest(n)
  24. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Reactive Relational Database Connectivity •

    A specification designed from the ground up for reactive programming • End to end reactive and non-blocking
  25. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Design Principles • Embrace Reactive

    Types and Patterns • Non-blocking, all the way to the database • Documented specification • Shrink the driver SPI • Enable multiple "humane" APIs
  26. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Drivers • H2 • Google

    Cloud Spanner • MySQL • Postgres • SQL Server • SAP Hana
  27. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 ConnectionFactory Transactions • Publisher<Void> beginTransaction();

    • Publisher<Void> commitTransaction(); • Publisher<Void> rollbackTransaction();
  28. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Spring Framework 5.2 • Reactive

    Transaction Manager SPI • Support for @Transactional • TransactionalOperator
  29. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 @Transactional • Method signature declares

    known reactive type • Requires ReactiveTransactionManager • Code path for reactive transaction management • Otherwise • Uses PlatformTransactionManager
  30. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 What is Neo4j? • A

    (property) graph database • An ecosystem • I work on Spring and Spring Data integration as part of the drivers team
  31. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 A Property Graph :Movie :Person

    :Person Nodes represents objects (Nouns) :DIRECTED :ACTED_IN role: someRole name: someName Relationships connect nodes and represent actions (verbs) Both nodes and relationships can have properties
  32. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 The whiteboard model IS the

    physical model • Just fill the abstract nodes with content
  33. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Neo4j 4.0 • Reactive Cypher

    engine • Reactive Driver available (Beta) • Spring Data Neo4j RX (Public preview) • Imperative and reactive transaction managers
  34. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Building blocks of Neo4j driver

    Driver Top level object for all Neo4j interaction Session Logical context for sequence of transactions Transaction Unit of work Statement result Stream of records plus metadata
  35. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Building blocks of Neo4j driver

    Driver Session Connection pool Connection Connection Server A Server B spawns borrows owns owns to
  36. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Building blocks of Neo4j driver

    • Session is a logical concept over the connection • One ongoing transaction per session • Thus, the connection is bound to the transaction
  37. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Challenges • Driver shall provide

    a similar surface for various languages • Java • JavaScript • Python • .NET
  38. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 More challenges • Already two

    different programming models • Imperative / Blocking • Asynchronous (Pretty much the default in .NET, feels like Co- Routines in Kotlin / Lightweight Threads in Loom) • Supported for all languages • Reactive coming up • Java • JavaScript • .NET
  39. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 @Test void implicit() { //

    Aka "Auto-Commit" try (Session session = driver.session()) { long personId = session .run( "CREATE (a:Person {name: $name}) RETURN id(a) as id“, parameters("name", NAME)) .single().get("id").asLong(); } } Implicit
  40. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 @Test void txFunctions() { try

    (Session session = driver.session()) { long personId = session.writeTransaction( tx -> tx.run( "CREATE (a:Person {name: $name}) RETURN id(a) as id“, parameters("name", NAME)) .single().get("id").asLong()); } } Transactional functions
  41. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 @Test void explicit() { try

    (Session session = driver.session()) { Transaction tx = session.beginTransaction(TransactionConfig.builder().build()); long personId = tx.run( "CREATE (a:Person {name: $name}) RETURN id(a) as id“, parameters("name", NAME)) .single().get("id").asLong(); tx.commit(); } } Explicit
  42. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 RxSession session = driver.rxSession(); Publisher<RxTransaction>

    txPublisher = session.beginTransaction(); RxTransaction tx; RxStatementResult result = tx.run(); Publisher<Record> recordPublisher = result.records() Publisher<Void> commit = tx.commit(); Publisher<Void> rollback = tx.rollback(); Reactive surface
  43. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 @Test void implicit() { Flux<Long>

    personCreation = Flux.using( driver::rxSession, session -> session.run( "CREATE (a:Person {name: $name}) RETURN id(a) as id“, parameters("name", NAME) ).records(), RxSession::close ).map(r -> r.get("id").asLong()); StepVerifier.create(personCreation) .expectNextCount(1) .verifyComplete(); } Implicit
  44. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 @Test void txFunctions() { RxTransactionWork<Publisher<Long>>

    txFunction = tx -> Flux.from(tx.run( "CREATE (a:Person {name: $name}) RETURN id(a) as id", parameters("name", NAME)).records() ) .map(r -> r.get("id").asLong()); Flux<Long> personCreation = Flux.using(driver::rxSession, session -> session.writeTransaction(txFunction), RxSession::close); StepVerifier.create(personCreation) .expectNextCount(1) .verifyComplete(); } Transactional functions
  45. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 @Test void explicit() { Function<RxSession,

    Flux<Long>> actualWork = session -> Flux.usingWhen(session.beginTransaction(), // Yes, this looks pretty much like the `txFunction` in the example before tx -> tx.run( "CREATE (a:Person {name: $name}) RETURN id(a) as id", parameters("name", NAME) ).records(), RxTransaction::commit, // Success case (tx, e) -> tx.rollback(), // Error / exceptional case RxTransaction::commit // Cancelation ).map(r -> r.get("id").asLong()); Flux<Long> personCreation = Flux.using(driver::rxSession, actualWork, RxSession::close); } Explicit
  46. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 How does SDN/RX differ from

    previous SDN? • Modular design • Fundamental design differences (e.g. mutability, incremental mapping) • SDN/OGM layers are now combined • Transport functionality moved entirely to Java driver • Supports all "findBy*" methods • Query by Example • Integration with Spring 5.2’s Reactive Transaction Manager
  47. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 SDN/RX Spring Data Repositories Neo4j

    Template Neo4j Client Driver Neo4j „uses“ sdn-rx-spring-boot-starter neo4j-java-driver-spring-boot-starter provides provides Integration with Spring Transactions from here on SDN/RX components
  48. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 ReactiveNeo4jClient reactiveClient; Flux<Map<String, Object>> directorAndMovies

    = reactiveClient .query( "MATCH (p:Person) - [:DIRECTED] -> (m:Movie {title: $title})," + " (p) - [:WROTE] -> (om:Movie) " + "WHERE p.name =~ $name " + " AND p.born < $someDate.year " + "RETURN p, om" ) .bind(LocalDate.of(1979, 9, 21)).to("someDate") .bindAll(Map.of("title", "The Matrix", "name", "Li.*")) .fetch() .all(); SDN/RX how does it look?
  49. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 @Service public class MovieService {

    @Transactional public Flux<MovieEntity> doSomething() { // Return type indicating reactive or imperative return Flux.empty(); } void doSomethingElse() { ReactiveNeo4jTemplate reactiveTemplate; TransactionalOperator transactionalOperator; Flux<MovieEntity> movies = transactionalOperator.transactional( reactiveTemplate.findAll(MovieEntity.class) ); } } And transactions?
  50. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 ReactiveNeo4jTransactionManager • Uses explicit reactive

    transactions • Uses the driver instance as lookup for the transactional resource • Stores both the session and transaction belonging to that session into resource holder
  51. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 ReactiveNeo4jTransactionManager • Basically this, in

    a Spring way: Function<RxSession, Flux<Long>> actualWork = session -> Flux.usingWhen( session.beginTransaction(), tx -> Flux.just(1L), RxTransaction::commit, (tx, e) -> tx.rollback(), RxTransaction::commit); Flux<Long> completeUnitOfWork = Flux.using(driver::rxSession, actualWork, RxSession::close)
  52. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 ReactiveNeo4jTransactionManager • Basically this, in

    a Spring way: • Session acquisition Function<RxSession, Flux<Long>> actualWork = session -> Flux.usingWhen( session.beginTransaction(), tx -> Flux.just(1L), RxTransaction::commit, (tx, e) -> tx.rollback(), RxTransaction::commit); Flux<Long> completeUnitOfWork = Flux.using(driver::rxSession, actualWork, RxSession::close)
  53. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 ReactiveNeo4jTransactionManager • Basically this, in

    a Spring way: • Session acquisition • Starting transaction Function<RxSession, Flux<Long>> actualWork = session -> Flux.usingWhen( session.beginTransaction(), tx -> Flux.just(1L), RxTransaction::commit, (tx, e) -> tx.rollback(), RxTransaction::commit); Flux<Long> completeUnitOfWork = Flux.using(driver::rxSession, actualWork, RxSession::close)
  54. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 ReactiveNeo4jTransactionManager • Basically this, in

    a Spring way: • Session acquisition • Starting transaction • Executing stuff Function<RxSession, Flux<Long>> actualWork = session -> Flux.usingWhen( session.beginTransaction(), tx -> Flux.just(1L), RxTransaction::commit, (tx, e) -> tx.rollback(), RxTransaction::commit); Flux<Long> completeUnitOfWork = Flux.using(driver::rxSession, actualWork, RxSession::close)
  55. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 ReactiveNeo4jTransactionManager • Basically this, in

    a Spring way: • Session acquisition • Starting transaction • Executing stuff • Commit or rollback Function<RxSession, Flux<Long>> actualWork = session -> Flux.usingWhen( session.beginTransaction(), tx -> Flux.just(1L), RxTransaction::commit, (tx, e) -> tx.rollback(), RxTransaction::commit); Flux<Long> completeUnitOfWork = Flux.using(driver::rxSession, actualWork, RxSession::close)
  56. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 ReactiveNeo4jTransactionManager • Basically this, in

    a Spring way: • Session acquisition • Starting transaction • Executing stuff • Commit or rollback • Closing session Function<RxSession, Flux<Long>> actualWork = session -> Flux.usingWhen( session.beginTransaction(), tx -> Flux.just(1L), RxTransaction::commit, (tx, e) -> tx.rollback(), RxTransaction::commit); Flux<Long> completeUnitOfWork = Flux.using(driver::rxSession, actualWork, RxSession::close)
  57. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Neo4j • https://neo4j.com/download/ • Neo4j

    Desktop (Analyst centric) • Neo4j Server (Community and Enterprise Edition) Community Edition: GPLv3 Enterprise Edition: Proprietary
  58. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Neo4j Datasets • Preconfigured instance

    with several different datasets https://neo4j.com/sandbox-v2/ • Neo4j Graph Gists, Example Models and Cypher Queries https://neo4j.com/graphgists/ • Panama papers and other offshore leaks https://offshoreleaks.icij.org/
  59. #Devoxx #ReactiveProgramming @mp911de and @rotnroll666 Spring related • Neo4j Driver

    Spring Boot Starter https://github.com/neo4j/neo4j-java-driver-spring-boot-starter • Spring Data Neo4j⚡RX https://github.com/neo4j/sdn-rx/