How to write transactional spout for storm trident -

i have written demo trident. in demo, want generate series batches of data in transaction spout , persistentaggregate operation them.

the normal processing this:

step1: txid=1, {"aaa 3", "bbb 2"}==>persist db(ok)

step2: txid=2, {"ccc 6", "ddd 7"}==>persist db(ok)

in case operation of persist db abnormal in step1, suppose process should be

step1: txid=1, {"aaa 3", "bbb 2"}==>persist db(ng)

step2: txid=1, {"aaa 3", "bbb 2"}==>persist db(ok)

but test result following:

step1: txid=1, {"aaa 3", "bbb 2"}==>persist db(ng)

step2: txid=1, {"ccc 6", "ddd 7"}==>persist db(ok)

i think process not correct.but not know why happen. have checked zookeeper, in path "meta/1, value null.

i wrote transactional spout myself implement interface itridentspout, think might problem spout have written.

anyone can give me advice? or give me implemented transactional spout example me reference.

thanks!