Java agent to trace rxjava subscription.

Details

Trace rxjava operators using unbounded subscription with Long#MAX_VALUE - they use ERROR/DROP/BUFFER overflow strategies to manage backpressure and in effect application can fail or loose data.

There is a group of applications which should run Subscription#request with manageable value after data is processed to keep memory under control and to not lose data.

This agent uses ByteBuddy and will instrument request method in any class implementing org.reactivestreams.Subscription, then log warning if it is executed with argument bigger than 10_000.

• Threshold can be changed using Java property, eg -Drxjava.agent.subscription.request-logging-threshold=500
• Warning includes stacktrace to identify which rxjava operator executed Subscription#request method. Stacktrace can be switched to normal warning with using a property -Drxjava.agent.subscription.log-stacktrace=false

Build

Use gradle to build agent JAR and run tests. Agent jar uses -all suffix and contains code running instrumentation and ByteBuddy code in ShadowJar. Agent jar will be placed in build/libs/rxjava-agent-1.0-SNAPSHOT-all.jar.

./gradlew build

Use agent directly

Use -javaagent argument to enable this Java agent on your application or tests. For example:

java -javaagent:../rxjava-agent/build/libs/rxjava-agent-1.0-SNAPSHOT-all.jar ...

# OR
java -Drxjava.agent.subscription.request-logging-threshold=500 \
  -javaagent:../rxjava-agent/build/libs/rxjava-agent-1.0-SNAPSHOT-all.jar ...

Use agent in Gradle tests

Publish the agent to mavenLocal repository

./gradlew publishToMavenLocal

Update Gradle config for project which will use the rxjava-agent to attach it when running tests

repositories {
  // ...
  mavenLocal()
}

val rxjavaAgent = configurations.create("rxjavaAgent")

dependencies {
    // ...
    rxjavaAgent("pl.kodstark:rxjava-agent:1.0.0-SNAPSHOT:all") { isTransitive = false }
}

tasks.withType<Test> {
    // ...
    jvmArgs = listOf(
        "-javaagent:${rxjavaAgent.asPath}"
    )
}