34
34
import io .streamnative .pulsar .handlers .kop .storage .PartitionLog ;
35
35
import io .streamnative .pulsar .handlers .kop .storage .ProducerStateManager ;
36
36
import io .streamnative .pulsar .handlers .kop .storage .ProducerStateManagerSnapshot ;
37
- import io .streamnative .pulsar .handlers .kop .storage .ProducerStateManagerSnapshotBuffer ;
38
37
import io .streamnative .pulsar .handlers .kop .storage .TxnMetadata ;
39
38
import java .lang .reflect .Method ;
40
39
import java .time .Duration ;
85
84
import org .apache .kafka .common .serialization .StringSerializer ;
86
85
import org .apache .pulsar .common .naming .TopicName ;
87
86
import org .awaitility .Awaitility ;
88
- import org .mockito .Mockito ;
89
87
import org .testng .Assert ;
90
88
import org .testng .annotations .AfterClass ;
91
89
import org .testng .annotations .AfterMethod ;
@@ -1042,7 +1040,7 @@ public void testPurgeAbortedTx(boolean takeSnapshotBeforeRecovery) throws Except
1042
1040
}
1043
1041
1044
1042
@ Test (timeOut = 10_000 )
1045
- public void testAbortedPurgeIntervalConfiguration () throws Exception {
1043
+ public void testAbortedTxnIndexPurgeIntervalConfiguration () throws Exception {
1046
1044
Class <ProducerStateManager > clazz = ProducerStateManager .class ;
1047
1045
Method maybePurgeMethod = clazz .getDeclaredMethod ("maybePurgeAbortedTx" );
1048
1046
maybePurgeMethod .setAccessible (true );
@@ -1069,7 +1067,7 @@ private ProducerStateManager buildProducerStateManager(Method updateAbortedTxnsP
1069
1067
ProducerStateManager producerStateManager = new ProducerStateManager (
1070
1068
"aborted-txn-index-purge-interval-test-" + RandomStringUtils .randomAlphanumeric (5 ),
1071
1069
UUID .randomUUID ().toString (),
1072
- Mockito . mock ( ProducerStateManagerSnapshotBuffer . class ) ,
1070
+ null ,
1073
1071
1000 * 30 ,
1074
1072
purgeAbortedTxnIntervalSec );
1075
1073
producerStateManager .updateMapEndOffset (100L );
0 commit comments