[GitHub] [pulsar] congbobo184 commented on a change in pull request #10564: [Transaction] Fix transaction buffer cache problem.

[GitBox <gi...@apache.org>]

congbobo184 commented on a change in pull request #10564:
URL: https://github.com/apache/pulsar/pull/10564#discussion_r631681349



##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
##########
@@ -833,4 +835,67 @@ public void transactionTimeoutTest() throws Exception {
         assertEquals(reReceiveMessage.getMessageId(), message.getMessageId());
 
     }
+
+    @DataProvider(name = "ackType")
+    public static Object[] ackType() {
+        return new Object[] {CommandAck.AckType.Cumulative, CommandAck.AckType.Individual};
+    }
+
+    @Test(dataProvider = "ackType")
+    public void txnTransactionRedeliverNullDispatcher(CommandAck.AckType ackType) throws Exception {
+        String topic = NAMESPACE1 + "/txnTransactionRedeliverNullDispatcher";
+        final String subName = "test";
+        @Cleanup
+        Consumer<byte[]> consumer = pulsarClient
+                .newConsumer()
+                .topic(topic)
+                .subscriptionName(subName)
+                .enableBatchIndexAcknowledgment(true)
+                .acknowledgmentGroupTime(0, TimeUnit.MILLISECONDS)
+                .subscribe();
+        Awaitility.await().until(consumer::isConnected);
+
+        @Cleanup
+        Producer<byte[]> producer = pulsarClient
+                .newProducer()
+                .topic(topic)
+                .sendTimeout(0, TimeUnit.SECONDS)
+                .enableBatching(false)
+                .create();
+
+
+        int messageCnt = 10;
+        for (int i = 0; i < messageCnt; i++) {
+            producer.send(("Hello Txn - " + i).getBytes(UTF_8));
+        }
+        Transaction txn = getTxn();
+        if (ackType == CommandAck.AckType.Individual) {
+            consumer.acknowledgeAsync(consumer.receive().getMessageId(), txn);
+        } else {
+            consumer.acknowledgeCumulativeAsync(consumer.receive().getMessageId(), txn);
+        }
+        topic = TopicName.get(topic).toString();
+        for (int i = 0; i < getPulsarServiceList().size(); i++) {
+
+            Field field = BrokerService.class.getDeclaredField("topics");
+            field.setAccessible(true);
+            ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics =
+                    (ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>>) field
+                            .get(getPulsarServiceList().get(i).getBrokerService());
+            CompletableFuture<Optional<Topic>> topicFuture = topics.get(topic);
+
+            if (topicFuture != null) {

Review comment:
       thanks for your review, add a flag to judge it.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org