Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

KAFKA-18035: TransactionsTest testBumpTransactionalEpochWithTV2Disabled failed on trunk #18451

Merged
merged 9 commits into from
Feb 11, 2025
Merged

KAFKA-18035: TransactionsTest testBumpTransactionalEpochWithTV2Disabled failed on trunk #18451

Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
94c80f3
Try to deflake some of the tests
jolshan Dec 18, 2024
c87077d
Change to get
jolshan Dec 19, 2024
75a150a
Merge branch 'trunk' of github.com:apache/kafka into kafka-18035
jolshan Jan 8, 2025
00a2853
logging back
jolshan Jan 8, 2025
c087a5c
more logging
jolshan Jan 10, 2025
ab25b1b
Ensure state transition
jolshan Jan 14, 2025
2194bcd
More elaborate waitUntil
jolshan Jan 15, 2025
2673e3b
Cleanups
jolshan Feb 11, 2025
a71dc3f
Merge branch 'trunk' of github.com:apache/kafka into KAFKA-18035
jolshan Feb 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 15 additions & 4 deletions core/src/test/scala/integration/kafka/api/TransactionsTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,8 @@ import kafka.utils.TestUtils.{consumeRecords, waitUntilTrue}
import kafka.utils.{TestInfoUtils, TestUtils}
import org.apache.kafka.clients.consumer._
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.{KafkaException, TopicPartition}
import org.apache.kafka.common.errors.{ConcurrentTransactionsException, InvalidProducerEpochException, ProducerFencedException, TimeoutException}
import org.apache.kafka.common.test.api.Flaky
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
import org.apache.kafka.coordinator.transaction.{TransactionLogConfig, TransactionStateManagerConfig}
import org.apache.kafka.server.config.{ReplicationConfigs, ServerConfigs, ServerLogConfigs}
Expand Down Expand Up @@ -693,7 +692,6 @@ class TransactionsTest extends IntegrationTestHarness {
assertThrows(classOf[IllegalStateException], () => producer.initTransactions())
}

@Flaky("KAFKA-18035")
@ParameterizedTest
@CsvSource(Array(
"kraft,classic,false",
Expand Down Expand Up @@ -733,6 +731,19 @@ class TransactionsTest extends IntegrationTestHarness {
restartDeadBrokers()

org.apache.kafka.test.TestUtils.assertFutureThrows(failedFuture, classOf[TimeoutException])
// Ensure the producer transitions to abortable_error state.
TestUtils.waitUntilTrue(() => {
var failed = false
try {
producer.send(TestUtils.producerRecordWithExpectedTransactionStatus(testTopic, 0, "3", "3", willBeCommitted = false))
} catch {
case e: Exception =>
if (e.isInstanceOf[KafkaException])
failed = true
}
failed
}, "The send request never failed as expected.")
assertThrows(classOf[KafkaException], () => producer.send(TestUtils.producerRecordWithExpectedTransactionStatus(testTopic, 0, "3", "3", willBeCommitted = false)))
producer.abortTransaction()

producer.beginTransaction()
Expand All @@ -755,7 +766,7 @@ class TransactionsTest extends IntegrationTestHarness {
producerStateEntry =
brokers(partitionLeader).logManager.getLog(new TopicPartition(testTopic, 0)).get.producerStateManager.activeProducers.get(producerId)
assertNotNull(producerStateEntry)
assertTrue(producerStateEntry.producerEpoch > initialProducerEpoch)
assertTrue(producerStateEntry.producerEpoch > initialProducerEpoch, "InitialProduceEpoch: " + initialProducerEpoch + " ProducerStateEntry: " + producerStateEntry)
} finally {
producer.close(Duration.ZERO)
}
Expand Down