Skip to content

Commit fcecca4

Browse files
authored
[fix][test] fix testEndTxnWhenCommittingOrAborting flaky test (apache#18318)
1 parent ccb7d82 commit fcecca4

3 files changed

Lines changed: 18 additions & 22 deletions

File tree

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,6 @@
3434
import org.apache.pulsar.common.api.proto.CommandTopicMigrated.ResourceType;
3535
import org.apache.pulsar.common.api.proto.ProtocolVersion;
3636
import org.apache.pulsar.common.api.proto.ServerError;
37-
import org.apache.pulsar.common.api.proto.TxnAction;
3837
import org.apache.pulsar.common.protocol.Commands;
3938
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
4039
import org.apache.pulsar.common.schema.SchemaInfo;
@@ -355,9 +354,6 @@ public void sendEndTxnErrorResponse(long requestId, TxnID txnID, ServerError err
355354
safeIntercept(command, cnx);
356355
ByteBuf outBuf = Commands.serializeWithSize(command);
357356
writeAndFlush(outBuf);
358-
if (this.interceptor != null) {
359-
this.interceptor.txnEnded(txnID.toString(), TxnAction.ABORT_VALUE);
360-
}
361357
}
362358

363359
@Override

pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java

Lines changed: 15 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -51,20 +51,20 @@
5151
@Slf4j
5252
public class CounterBrokerInterceptor implements BrokerInterceptor {
5353

54-
private AtomicInteger beforeSendCount = new AtomicInteger();
55-
private AtomicInteger beforeSendCountAtConsumerLevel = new AtomicInteger();
56-
private AtomicInteger count = new AtomicInteger();
57-
private AtomicInteger connectionCreationCount = new AtomicInteger();
58-
private AtomicInteger producerCount = new AtomicInteger();
59-
private AtomicInteger consumerCount = new AtomicInteger();
60-
private AtomicInteger messagePublishCount = new AtomicInteger();
61-
private AtomicInteger messageCount = new AtomicInteger();
62-
private AtomicInteger messageDispatchCount = new AtomicInteger();
63-
private AtomicInteger messageAckCount = new AtomicInteger();
64-
private AtomicInteger handleAckCount = new AtomicInteger();
65-
private AtomicInteger txnCount = new AtomicInteger();
66-
private AtomicInteger committedTxnCount = new AtomicInteger();
67-
private AtomicInteger abortedTxnCount = new AtomicInteger();
54+
private final AtomicInteger beforeSendCount = new AtomicInteger();
55+
private final AtomicInteger beforeSendCountAtConsumerLevel = new AtomicInteger();
56+
private final AtomicInteger count = new AtomicInteger();
57+
private final AtomicInteger connectionCreationCount = new AtomicInteger();
58+
private final AtomicInteger producerCount = new AtomicInteger();
59+
private final AtomicInteger consumerCount = new AtomicInteger();
60+
private final AtomicInteger messagePublishCount = new AtomicInteger();
61+
private final AtomicInteger messageCount = new AtomicInteger();
62+
private final AtomicInteger messageDispatchCount = new AtomicInteger();
63+
private final AtomicInteger messageAckCount = new AtomicInteger();
64+
private final AtomicInteger handleAckCount = new AtomicInteger();
65+
private final AtomicInteger txnCount = new AtomicInteger();
66+
private final AtomicInteger committedTxnCount = new AtomicInteger();
67+
private final AtomicInteger abortedTxnCount = new AtomicInteger();
6868

6969
public void reset() {
7070
beforeSendCount.set(0);
@@ -81,7 +81,7 @@ public void reset() {
8181
abortedTxnCount.set(0);
8282
}
8383

84-
private List<ResponseEvent> responseList = new ArrayList<>();
84+
private final List<ResponseEvent> responseList = new ArrayList<>();
8585

8686
@Data
8787
@AllArgsConstructor

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -920,11 +920,11 @@ public void testEndTxnWhenCommittingOrAborting() throws Exception {
920920
field.set(abortTxn, TransactionImpl.State.ABORTING);
921921

922922

923-
assertEquals(((CounterBrokerInterceptor)listener).getTxnCount(),2);
923+
Awaitility.await().untilAsserted(() -> assertEquals(listener.getTxnCount(),2));
924924
abortTxn.abort().get();
925-
assertEquals(((CounterBrokerInterceptor)listener).getAbortedTxnCount(),1);
925+
Awaitility.await().untilAsserted(() -> assertEquals(listener.getAbortedTxnCount(),1));
926926
commitTxn.commit().get();
927-
assertEquals(((CounterBrokerInterceptor)listener).getCommittedTxnCount(),1);
927+
Awaitility.await().untilAsserted(() -> assertEquals(listener.getCommittedTxnCount(),1));
928928
}
929929

930930
@Test

0 commit comments

Comments
 (0)