enable pulsar txn coordinator

Author: dao-jun

instrumentation/pulsar/pulsar-2.8/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/AbstractPulsarClientTest.java

@@ -88,6 +88,8 @@ static void beforeAll() throws PulsarClientException {
     pulsar =
         new PulsarContainer(DEFAULT_IMAGE_NAME)
             .withEnv("PULSAR_MEM", "-Xmx128m")
+            .withEnv("PULSAR_PREFIX_transactionCoordinatorEnabled", "true")
+            .withEnv("PULSAR_PREFIX_transactionMetadataStoreProviderClassName", "org.apache.pulsar.transaction.coordinator.impl.InMemTransactionMetadataStoreProvider")
             .withLogConsumer(new Slf4jLogConsumer(logger))
             .withStartupTimeout(Duration.ofMinutes(2));
     pulsar.start();

instrumentation/pulsar/pulsar-2.8/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/PulsarClientTest.java

@@ -677,12 +677,13 @@ void testConsumePartitionedTopicUsingBatchReceive() throws Exception {
   void testSendMessageWithTxn() throws Exception {
     String topic = "persistent://public/default/testSendMessageWithTxn";
     admin.topics().createNonPartitionedTopic(topic);
-    producer = client.newProducer(Schema.STRING).topic(topic).enableBatching(false).create();
+    producer = client.newProducer(Schema.STRING).topic(topic)
+        .sendTimeout(0, TimeUnit.SECONDS).enableBatching(false).create();
     Transaction txn =
         client.newTransaction().withTransactionTimeout(5, TimeUnit.SECONDS).build().get();
+    txn.commit();
 
     testing.runWithSpan("parent1", () -> producer.newMessage(txn).value("test1").send());
-
     testing.waitAndAssertTraces(
         trace ->
             trace.hasSpansSatisfyingExactly(