freef4ll
commented
2 years ago This is reproducible on 22.2.6.
Create topics:
export REDPANDA_BROKERS="linux:9092"
export REDPANDA_API_ADMIN_ADDRS="linux:9644"
rpk cluster config set enable_transactions true
rpk topic create txn_input --partitions 6 #-replicas 3
rpk topic create txn_output --partitions 6 #-replicas 3Build librdkafka and apply transaction_id.patch.txt :
git clone https://github.com/edenhill/librdkafka.git
cd librdkafka
patch < transaction_id.patch.txt
./configure && makeA small wrapper for transactions to save logs:
cat run-transaction.sh
#!/bin/bash
if [ "$#" -ne 1 ]; then
echo "usage: $0 transcation_id"
exit -1
fi
transaction_id="${1}"
log_file="transactions_${transaction_id}_$(date +'%Y%m%d_%H%M')"
export TZ=UTC
./examples/transactions $REDPANDA_BROKERS txn_input txn_output ${transaction_id} 2>&1 | tee -a ${log_file}
#| perl -pe 's/([\d]{10})/localtime $1/eg;' | tee -a ${log_file}Produce data into processing topic:
./examples/rdkafka_performance -b ${REDPANDA_BROKERS} -P -t txn_input -m "1 " -s 1000 -r 1000Launch single transaction processor:
bash run-transaction.sh id_first_$$Once data has been started to be processed, start the 2nd instance which will cause the first instance to abort.
bash run-transaction.sh id_second_$$Trace logs from Redpanda:
TRACE 2022-10-27 16:03:48,789 [shard 0] kafka - request_context.h:160 - [192.168.66.1:53790] sending 1:fetch response {throttle_time_ms=0 error_code={ error_code: none [0] } session_id=0 topics={{name={txn_input} partitions={{partition_index=4 error_code={ error_code: none [0] } high_watermark=54737 last_stable_offset=54737 log_start_offset=0 aborted={nullopt} preferred_read_replica=-1 records={{size 1602327}}}, {partition_index=5 error_code={ error_code: none [0] } high_watermark=53383 last_stable_offset=53383 log_start_offset=0 aborted={nullopt} preferred_read_replica=-1 records={{size 1606371}}}, {partition_index=3 error_code={ error_code: none [0] } high_watermark=53148 last_stable_offset=53148 log_start_offset=0 aborted={nullopt} preferred_read_replica=-1 records={{size 1589182}}}}}}}
TRACE 2022-10-27 16:03:48,791 [shard 0] tx - rm_group_frontend.cc:263 - sending name:begin_group_tx, group_id:{librdkafka_transactions_example_group}, pid:{producer_identity: id=11004, epoch=0}, tx_seq:1884, ec:cluster::tx_errc:0, etag:25
TRACE 2022-10-27 16:03:48,792 [shard 0] tx - tx_gateway_frontend.cc:49 - released_lock name:add_offsets_to_tx, tx_id:{id_first_52628}
TRACE 2022-10-27 16:03:48,792 [shard 0] kafka - request_context.h:160 - [192.168.66.1:53786] sending 25:add_offsets_to_txn response {throttle_time_ms=0 error_code={ error_code: none [0] }}
TRACE 2022-10-27 16:03:48,796 [shard 0] kafka - requests.cc:91 - [192.168.66.1:53786] processing name:add_partitions_to_txn, key:24, version:0 for rdkafka, mem_units: 8134
TRACE 2022-10-27 16:03:48,796 [shard 0] tx - tx_gateway_frontend.cc:45 - got_lock name:add_partition_to_tx, tx_id:{id_first_52628}
TRACE 2022-10-27 16:03:48,796 [shard 0] tx - rm_partition_frontend.cc:247 - processing name:begin_tx, ntp:{kafka/txn_output/5}, pid:{producer_identity: id=11004, epoch=0}, tx_seq:1884
ERROR 2022-10-27 16:03:48,796 [shard 0] cluster - rm_stm.cc:364 - there is already an ongoing transaction within {producer_identity: id=11004, epoch=0} session
WARN 2022-10-27 16:03:48,796 [shard 0] tx - rm_partition_frontend.cc:303 - rm_stm::begin_tx({kafka/txn_output/5},...) failed with cluster::tx_errc:16
TRACE 2022-10-27 16:03:48,796 [shard 0] tx - rm_partition_frontend.cc:256 - sending name:begin_tx, ntp:{kafka/txn_output/5}, pid:{producer_identity: id=11004, epoch=0}, tx_seq:1884, ec:cluster::tx_errc:16, etag:-9223372036854775808
WARN 2022-10-27 16:03:48,796 [shard 0] tx - rm_partition_frontend.cc:149 - local execution of begin_tx({kafka/txn_output/5},...) failed with cluster::tx_errc:16
WARN 2022-10-27 16:03:48,796 [shard 0] tx - tx_gateway_frontend.cc:981 - begin_tx({kafka/txn_output/5},...) failed with cluster::tx_errc:16
TRACE 2022-10-27 16:03:48,796 [shard 0] tx - tx_gateway_frontend.cc:49 - released_lock name:add_partition_to_tx, tx_id:{id_first_52628}
TRACE 2022-10-27 16:03:48,796 [shard 0] kafka - request_context.h:160 - [192.168.66.1:53786] sending 24:add_partitions_to_txn response {throttle_time_ms=0 results={{name={txn_output} results={{partition_index=5 error_code={ error_code: invalid_txn_state [48] }}}}}}
TRACE 2022-10-27 16:03:48,797 [shard 0] kafka - requests.cc:91 - [192.168.66.1:53791] processing name:txn_offset_commit, key:28, version:3 for rdkafka, mem_units: 8326
TRACE 2022-10-27 16:03:48,797 [shard 0] kafka - txn_offset_commit.cc:86 - Handling request {transactional_id=id_first_52628 group_id={librdkafka_transactions_example_group} producer_id=11004 producer_epoch=0 generation_id=88 member_id={rdkafka-6970344e-9228-4f0c-bf63-adcc3373d66b} group_instance_id={nullopt} topics={{name={txn_input} partitions={{partition_index=3 committed_offset=49979 committed_leader_epoch=-1 committed_metadata={}}}}}}
TRACE 2022-10-27 16:03:48,797 [shard 0] kafka - request_context.h:160 - [192.168.66.1:53791] sending 28:txn_offset_commit response {throttle_time_ms=0 topics={{name={txn_input} partitions={{partition_index=3 error_code={ error_code: none [0] }}}}}}
TRACE 2022-10-27 16:03:48,811 [shard 0] exception - Throw exception at:
0x43c1827 0x41089f3 /opt/redpanda/lib/libc++abi.so.1+0x2ca97 0x41f10d3 0x41f11b7 0x41c44cb 0x41c732f 0x41c5417 0x40f85db 0x40f6dd7 0x17c662f 0x4451567 /opt/redpanda/lib/libc.so.6+0x24a9b 0x17c3637
--------
seastar::continuation<seastar::internal::promise_base_with_type<seastar::temporary_buffer<char> >, seastar::reactor::do_read_some(seastar::pollable_fd_state&, seastar::internal::buffer_allocator*)::$_34, seastar::future<seastar::temporary_buffer<char> > seastar::future<void>::then_impl_nrvo<seastar::reactor::do_read_some(seastar::pollable_fd_state&, seastar::internal::buffer_allocator*)::$_34, seastar::future<seastar::temporary_buffer<char> > >(seastar::reactor::do_read_some(seastar::pollable_fd_state&, seastar::internal::buffer_allocator*)::$_34&&)::'lambda'(seastar::internal::promise_base_with_type<seastar::temporary_buffer<char> >&&, seastar::reactor::do_read_some(seastar::pollable_fd_state&, seastar::internal::buffer_allocator*)::$_34&, seastar::future_state<seastar::internal::monostate>&&), void>
--------
seastar::continuation<seastar::internal::promise_base_with_type<seastar::temporary_buffer<char> >, seastar::net::posix_data_source_impl::get()::$_2, seastar::future<seastar::temporary_buffer<char> > seastar::future<seastar::temporary_buffer<char> >::then_impl_nrvo<seastar::net::posix_data_source_impl::get()::$_2, seastar::future<seastar::temporary_buffer<char> > >(seastar::net::posix_data_source_impl::get()::$_2&&)::'lambda'(seastar::internal::promise_base_with_type<seastar::temporary_buffer<char> >&&, seastar::net::posix_data_source_impl::get()::$_2&, seastar::future_state<seastar::temporary_buffer<char> >&&), seastar::temporary_buffer<char> >
--------
seastar::continuation<seastar::internal::promise_base_with_type<seastar::temporary_buffer<char> >, seastar::input_stream<char>::read_exactly(unsigned long)::'lambda'(auto), seastar::future<seastar::temporary_buffer<char> > seastar::future<seastar::temporary_buffer<char> >::then_impl_nrvo<seastar::input_stream<char>::read_exactly(unsigned long)::'lambda'(auto), seastar::future<seastar::temporary_buffer<char> > >(auto&&)::'lambda'(seastar::internal::promise_base_with_type<seastar::temporary_buffer<char> >&&, seastar::input_stream<char>::read_exactly(unsigned long)::'lambda'(auto)&, seastar::future_state<seastar::temporary_buffer<char> >&&), seastar::temporary_buffer<char> >
--------
seastar::internal::coroutine_traits_base<std::__1::optional<unsigned long> >::promise_type
--------
seastar::continuation<seastar::internal::promise_base_with_type<void>, kafka::connection_context::process_one_request()::$_0, seastar::future<void> seastar::future<std::__1::optional<unsigned long> >::then_impl_nrvo<kafka::connection_context::process_one_request()::$_0, seastar::future<void> >(kafka::connection_context::process_one_request()::$_0&&)::'lambda'(seastar::internal::promise_base_with_type<void>&&, kafka::connection_context::process_one_request()::$_0&, seastar::future_state<std::__1::optional<unsigned long> >&&), std::__1::optional<unsigned long> >
--------
seastar::internal::do_until_state<kafka::protocol::apply(net::server::resources)::$_4, kafka::protocol::apply(net::server::resources)::$_3>
--------
seastar::continuation<seastar::internal::promise_base_with_type<void>, seastar::future<void> seastar::future<void>::handle_exception<kafka::protocol::apply(net::server::resources)::$_5>(kafka::protocol::apply(net::server::resources)::$_5&&)::'lambda'(kafka::protocol::apply(net::server::resources)::$_5&&), seastar::futurize<kafka::protocol::apply(net::server::resources)::$_5>::type seastar::future<void>::then_wrapped_nrvo<seastar::future<void>, seastar::future<void> seastar::future<void>::handle_exception<kafka::protocol::apply(net::server::resources)::$_5>(kafka::protocol::apply(net::server::resources)::$_5&&)::'lambda'(kafka::protocol::apply(net::server::resources)::$_5&&)>(seastar::future<void> seastar::future<void>::handle_exception<kafka::protocol::apply(net::server::resources)::$_5>(kafka::protoco:Full trace log: rpk_debug.txt.gz
emaxerrno
bharathv
rystsov
dotnwat
piyushredpanda
mr-swifter
Version & Environment
Redpanda version: Redpanda v22.2.2 - b6d1d461c81519a97b5ad52292fd52e796f05816 Operating System: MacOS 12.5 librdkafka v/1.9.1 / v.1.9.2 Docker: Desktop 4.11.0, Engine: 20.10.17
What went wrong?
I have two applications, consuming data within the same consumer group from topic A (6 partitions) and write data to topic B. All done in transactions (committing offset to A + writing data to B).
I start first application and it works fine, it gets all partitions assigned to the itself and process data normally. When I start second application in actually kicks out the first one. What happens is first app gets abortable transaction error, so it tries to abort transaction. But transaction abort return unknown fatal error.
librdkafka logs from first application: first-app-with-redpanda.txt
transactional.ids are different for sure.What should have happened instead?
I tested the same with Kafka (v.2.13-3.2.1) and it seems it works normally, so transaction was successfully aborted. first-app-with-kafka.txt
How to reproduce the issue?
It can be reproduced with librdkafka transaction example code. A few changes require:
This basically to increate rate of transactions, so second app most probably will come when first app in. the middle of transaction
Additional information
I found the https://github.com/redpanda-data/redpanda/issues/3142, but I'm not sure it's the same, since leadership seems is not changing at that time, only rebalancing when new consumer joins.