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

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

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



##########
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:
       is it possible that this variable is `null` ?
   
   in case it is `null` this test is not testing your fix.




-- 
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