Activity interface which extends an interface with some method using generic types always throws a ClassCastException when said method is called in a Workflow

Expected Behavior

I would expect that an activity interface that extends some other interface with generic types and a method that uses those generic types would have that inherited method be executable as an activity method within a workflow.

Actual Behavior

A method that's inherited and uses generics in an activity interface always throws a ClassCastException when called within a workflow.

An example of the exception (using the repro code included in this issue) is. as follows:

java.lang.ClassCastException: class java.util.LinkedHashMap cannot be cast to class issue.repro.Foo (java.util.LinkedHashMap is in module java.base of loader 'bootstrap'; issue.repro.Foo is in unnamed module of loader 'app')
	at issue.repro.MyActivityImpl.run(ActivityIssueReproTest.kt:38)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at io.temporal.internal.activity.POJOActivityTaskHandler$POJOActivityInboundCallsInterceptor.execute(POJOActivityTaskHandler.java:214)
	at io.temporal.internal.activity.POJOActivityTaskHandler$POJOActivityImplementation.execute(POJOActivityTaskHandler.java:180)
	at io.temporal.internal.activity.POJOActivityTaskHandler.handle(POJOActivityTaskHandler.java:120)
	at io.temporal.internal.worker.LocalActivityWorker$TaskHandlerImpl.handleLocalActivity(LocalActivityWorker.java:235)
	at io.temporal.internal.worker.LocalActivityWorker$TaskHandlerImpl.handle(LocalActivityWorker.java:207)
	at io.temporal.internal.worker.LocalActivityWorker$TaskHandlerImpl.handle(LocalActivityWorker.java:196)
	at io.temporal.internal.worker.PollTaskExecutor.lambda$process$0(PollTaskExecutor.java:93)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)

Steps to Reproduce the Problem

The repro Kotlin code below uses the Temporal Java SDK (with the Kotlin library), and mockk and Kotest.

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.kotlin.registerKotlinModule
import io.kotest.core.spec.style.ShouldSpec
import io.kotest.matchers.shouldBe
import io.mockk.every
import io.mockk.mockk
import io.mockk.verify
import io.temporal.activity.ActivityInterface
import io.temporal.activity.LocalActivityOptions
import io.temporal.client.WorkflowClientOptions
import io.temporal.client.WorkflowOptions
import io.temporal.common.converter.DefaultDataConverter
import io.temporal.common.converter.JacksonJsonPayloadConverter
import io.temporal.internal.logging.LoggerTag
import io.temporal.testing.TestEnvironmentOptions
import io.temporal.testing.TestWorkflowEnvironment
import io.temporal.worker.Worker
import io.temporal.worker.WorkerOptions
import io.temporal.worker.WorkflowImplementationOptions
import io.temporal.workflow.Workflow
import io.temporal.workflow.WorkflowInterface
import io.temporal.workflow.WorkflowMethod
import java.time.Duration

class Foo

class Bar

interface InterfaceWithGenerics<InputT : Any, OutputT : Any> {
    fun run(input: InputT): OutputT
}

@ActivityInterface
interface MyActivity : InterfaceWithGenerics<Foo, Bar>

class MyActivityImpl : MyActivity {
    override fun run(input: Foo): Bar {
        return Bar()
    }
}

@WorkflowInterface
interface MyWorkflow {
    @WorkflowMethod
    fun run(input: Foo): Bar
}

class MyWorkflowImpl : MyWorkflow {
    private val myActivity = Workflow.newLocalActivityStub(
        MyActivity::class.java,
        LocalActivityOptions { setStartToCloseTimeout(Duration.ofSeconds(10)) }
    )

    override fun run(input: Foo): Bar {
        return myActivity.run(input)
    }
}

class ActivityIssueReproTest : ShouldSpec({
    val myActivity = mockk<MyActivityImpl>()

    val testEnv: TestWorkflowEnvironment = run {
        val objectMapper: ObjectMapper = ObjectMapper().also { mapper ->
            mapper.registerKotlinModule()
        }
        val dataConverter = DefaultDataConverter(JacksonJsonPayloadConverter(objectMapper))
        val workflowClientOptions = WorkflowClientOptions.newBuilder()
            .setDataConverter(dataConverter)
            .build()
        val workflowTestEnvironmentOptions = TestEnvironmentOptions.newBuilder()
            .setWorkflowClientOptions(workflowClientOptions)
            .build()

        TestWorkflowEnvironment.newInstance(workflowTestEnvironmentOptions)
    }

    val worker: Worker = run {
        val workerOptions: WorkerOptions = WorkerOptions.newBuilder()
            .setDefaultDeadlockDetectionTimeout(5_000L)
            .validateAndBuildWithDefaults()

        testEnv.newWorker(LoggerTag.TASK_QUEUE, workerOptions).also { worker ->
            // Register the workflow implementation(s)
            worker.registerWorkflowImplementationTypes(
                WorkflowImplementationOptions { setFailWorkflowExceptionTypes(Throwable::class.java) },
                MyWorkflowImpl::class.java
            )
            // Register the activity implementation(s)
            worker.registerActivitiesImplementations(myActivity)
        }
    }

    val defaultWorkflowOptions = WorkflowOptions
        .newBuilder()
        .setTaskQueue(LoggerTag.TASK_QUEUE)
        .build()

    beforeTest {
        testEnv.start()
    }

    should("repro the Temporal issue we're seeing") {
        val response = Bar()
        every { myActivity.run(any()) } returns response

        val myWorkflow = testEnv.workflowClient.newWorkflowStub(
            MyWorkflow::class.java,
            defaultWorkflowOptions
        )

        val foo = Foo()
        val result = myWorkflow.run(foo)

        result shouldBe response
        verify {
            myActivity.run(foo)
        }
    }

    afterTest {
        testEnv.close()
    }
})

Specifications

• Version: 1.8.0
• Platform: macOS

Start with the POJOActivityTaskHandler stack frames and the Kotlin ActivityIssueReproTest example. Trace how the inherited generic run method is discovered and how its argument is decoded, then verify the repro no longer casts a LinkedHashMap to Foo and the activity completes successfully.