Skip to content

fix(inkless:ci): green the upstream test suite (diskless test fixes + flaky quarantine) #700

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

Open
jeqo wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix(inkless:ci): green the upstream test suite (diskless test fixes + flaky quarantine) #700

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1045633
fix(inkless:ci): update upstream tests for diskless additions
jeqo Jul 14, 2026
61fe026
test(inkless:ci): quarantine known-flaky upstream integration tests w…
jeqo Jul 15, 2026
7607750
fix(inkless:ci): make the flaky test job report-only so CI is trustable
jeqo Jul 15, 2026
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
7 changes: 6 additions & 1 deletion .github/scripts/junit.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -347,6 +347,11 @@ def split_report_path(base_path: str, report_path: str) -> Tuple[str, str]:
exit_code = get_env("GRADLE_TEST_EXIT_CODE", int)
junit_report_url = get_env("JUNIT_REPORT_URL")
thread_dump_url = get_env("THREAD_DUMP_URL")
# Inkless: in the flaky/quarantine test job, tests fail by definition (they are @Flaky-tagged
# and known unreliable). Their failures must not fail the job, otherwise the workflow shows a
# red signal even when nothing at the Kafka coverage level is broken. Real infrastructure
# problems (timeout=124, thread dumps, missing exit code) still fail the job below.
run_flaky = (get_env("RUN_FLAKY") or "").lower() == "true"

if exit_code is None:
failure_messages.append("Missing required GRADLE_TEST_EXIT_CODE environment variable. Failing this script.")
Expand All @@ -357,7 +362,7 @@ def split_report_path(base_path: str, report_path: str) -> Tuple[str, str]:
# this script. If any task fails due to timeout, we want to fail the overall build since it will not
# include all the test results
failure_messages.append(f"Gradle task had a timeout. Failing this script. These are partial results!")
elif exit_code > 0:
elif exit_code > 0 and not run_flaky:
failure_messages.append(f"Gradle task had a failure exit code. Failing this script.")

if thread_dump_url:
Expand Down
3 changes: 3 additions & 0 deletions .github/workflows/build.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -273,6 +273,9 @@ jobs:
JUNIT_REPORT_URL: ${{ steps.archive-junit-html.outputs.artifact-url }}
THREAD_DUMP_URL: ${{ steps.archive-thread-dump.outputs.artifact-url }}
GRADLE_TEST_EXIT_CODE: ${{ steps.junit-test.outputs.gradle-exitcode }}
# Inkless: flaky-test failures in the flaky job are expected and must not fail the job,
# so the workflow only goes red when something at the Kafka coverage level actually breaks.
RUN_FLAKY: ${{ matrix.run-flaky }}
run: |
python .github/scripts/junit.py \
--path build/junit-xml >> $GITHUB_STEP_SUMMARY
Expand Down
2 changes: 2 additions & 0 deletions build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3027,6 +3027,7 @@ project(':streams') {
testCompileOnly libs.bndlib

testImplementation project(':clients').sourceSets.test.output
testImplementation project(':test-common:test-common-util')
testImplementation libs.jacksonDataformatYaml
testImplementation libs.junitJupiter
testImplementation libs.bcpkix
Expand Down Expand Up @@ -4192,6 +4193,7 @@ project(':connect:mirror') {
testImplementation project(':connect:runtime').sourceSets.test.output
testImplementation project(':core')
testImplementation project(':test-common:test-common-runtime')
testImplementation project(':test-common:test-common-util')
testImplementation project(':server')
testImplementation project(':server-common')

Expand Down
2 changes: 2 additions & 0 deletions ...-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.Flaky;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.coordinator.group.GroupConfig;
Expand Down Expand Up @@ -1716,6 +1717,7 @@ public void onComplete(Map<TopicIdPartition, Set<Long>> offsetsMap, Exception ex
* causes it to throw InterruptException
*/
@ClusterTest
@Flaky(value = "KAFKA-19961", comment = "Flaky ShareConsumerTest#testPollThrowsInterruptExceptionIfInterrupted.")
public void testPollThrowsInterruptExceptionIfInterrupted() {
alterShareAutoOffsetReset("group1", "earliest");
try (ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer("group1")) {
Expand Down
6 changes: 4 additions & 2 deletions clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,8 +58,10 @@ public void testUnsupportedVersionsToString() {

@Test
public void testUnknownApiVersionsToString() {
NodeApiVersions versions = NodeApiVersions.create((short) 337, (short) 0, (short) 1);
assertTrue(versions.toString().endsWith("UNKNOWN(337): 0 to 1)"));
// Inkless: use an id above all real ApiKeys (incl. diskless keys 500/501) so that the
// unknown key sorts last in NodeApiVersions.toString()'s ascending-by-id output.
NodeApiVersions versions = NodeApiVersions.create(Short.MAX_VALUE, (short) 0, (short) 1);
assertTrue(versions.toString().endsWith("UNKNOWN(" + Short.MAX_VALUE + "): 0 to 1)"));
}

@Test
Expand Down
5 changes: 5 additions & 0 deletions ...java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.test.api.Flaky;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Timer;
Expand Down Expand Up @@ -280,6 +281,7 @@ public void shutdownClusters() throws Exception {
}

@Test
@Flaky(value = "KAFKA-16488", comment = "Chronically flaky MM2 integration test across all subclasses; see also KAFKA-12566.")

@viktorsomogyi viktorsomogyi Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As far as I know this annotation can't be inherited as the annotation processor doesn't walk the method hierarchy. Did your test flag this as flaky though?

public void testReplication() throws Exception {
produceMessages(primaryProducer, "test-topic-1");
String backupTopic1 = remoteTopicName("test-topic-1", PRIMARY_CLUSTER_ALIAS);
Expand Down Expand Up @@ -816,6 +818,7 @@ private Map<TopicPartition, OffsetAndMetadata> partialOffsets(ConsumerRecords<by
}

@Test
@Flaky(value = "KAFKA-15945", comment = "Chronically flaky MM2 integration test across all subclasses; see also KAFKA-15523, KAFKA-14971.")
public void testSyncTopicConfigs() throws InterruptedException {
mm2Config = new MirrorMakerConfig(mm2Props);

Expand Down Expand Up @@ -859,6 +862,7 @@ public void testSyncTopicConfigs() throws InterruptedException {
}

@Test
@Flaky(value = "KAFKA-15926", comment = "Chronically flaky MM2 integration test across all subclasses; open tickets KAFKA-15926 (SSL), KAFKA-15927 (ExactlyOnce).")
public void testReplicateSourceDefault() throws Exception {
mm2Props.put(DefaultConfigPropertyFilter.USE_DEFAULTS_FROM, "source");
mm2Config = new MirrorMakerConfig(mm2Props);
Expand Down Expand Up @@ -901,6 +905,7 @@ public void testReplicateSourceDefault() throws Exception {
}

@Test
@Flaky(value = "KAFKA-15926", comment = "No dedicated ticket; same MM2 replication-harness flakiness as testReplicateSourceDefault (KAFKA-15926/KAFKA-15927).")
public void testReplicateTargetDefault() throws Exception {
mm2Config = new MirrorMakerConfig(mm2Props);

Expand Down
2 changes: 2 additions & 0 deletions ...me/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.test.api.Flaky;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
import org.apache.kafka.connect.storage.StringConverter;
import org.apache.kafka.connect.util.clusters.EmbeddedConnectStandalone;
Expand Down Expand Up @@ -80,6 +81,7 @@ public void close() {
}

@Test
@Flaky(value = "KAFKA-18952", comment = "Flaky in CI; upstream ticket resolved but still recurs in the fork.")
public void testMonitorableSinkConnectorAndTask() throws Exception {
connect.kafka().createTopic("test-topic");

Expand Down
11 changes: 11 additions & 0 deletions core/src/test/scala/integration/kafka/api/SaslOAuthBearerSslEndToEndAuthorizationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,22 @@
package kafka.api

import kafka.security.JaasTestUtils
import kafka.utils.TestInfoUtils
import org.apache.kafka.common.security.auth._
import org.apache.kafka.common.test.api.Flaky
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.MethodSource

class SaslOAuthBearerSslEndToEndAuthorizationTest extends SaslEndToEndAuthorizationTest {
override protected def kafkaClientSaslMechanism = "OAUTHBEARER"
override protected def kafkaServerSaslMechanisms = List(kafkaClientSaslMechanism)
override val clientPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, JaasTestUtils.KAFKA_OAUTH_BEARER_USER)
override val kafkaPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, JaasTestUtils.KAFKA_OAUTH_BEARER_ADMIN)

// Override solely to mark the OAUTHBEARER variant flaky; other SASL mechanisms keep upstream coverage.
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedGroupProtocolNames)
@MethodSource(Array("getTestGroupProtocolParametersAll"))
@Flaky(value = "KAFKA-9655", comment = "Flaky only for the OAUTHBEARER variant; non-critical for the fork, upstream covers the other mechanisms.")
override def testNoConsumeWithoutDescribeAclViaSubscribe(groupProtocol: String): Unit =
super.testNoConsumeWithoutDescribeAclViaSubscribe(groupProtocol)
}
11 changes: 9 additions & 2 deletions core/src/test/scala/unit/kafka/metrics/MetricsTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,8 +139,15 @@ class MetricsTest extends KafkaServerTestHarness with Logging {
createTopic(topic, 2)

// The broker metrics for all topics should be greedily registered
assertTrue(topicMetrics(None).nonEmpty, "General topic metrics don't exist")
assertEquals(brokers.head.brokerTopicStats.allTopicsStats.metricMapKeySet().size, topicMetrics(None).size)
val allTopicsMetrics = topicMetrics(None)
assertTrue(allTopicsMetrics.nonEmpty, "General topic metrics don't exist")
// Inkless: all-topics BytesIn/BytesOut meters are registered for both topicType=classic and
// topicType=diskless. The diskless variants share the same metric name as their classic
// counterparts (they differ only by tag), so they are not tracked as extra keys in
// metricMapKeySet() but do appear as separate JMX MBeans. Account for those duplicates.
val disklessAllTopicsMetrics = allTopicsMetrics.count(_.contains("topicType=diskless"))
assertEquals(brokers.head.brokerTopicStats.allTopicsStats.metricMapKeySet().size + disklessAllTopicsMetrics,
allTopicsMetrics.size)
assertEquals(0, brokers.head.brokerTopicStats.allTopicsStats.metricGaugeMap.size)
// topic metrics should be lazily registered
assertTrue(topicMetricGroups(topic).isEmpty, "Topic metrics aren't lazily registered")
Expand Down
3 changes: 2 additions & 1 deletion metadata/src/test/java/org/apache/kafka/image/node/TopicImageNodeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,8 @@ public void testChildPartitionId() {
assertEquals("PartitionRegistration(replicas=[2, 3, 4], " +
"directories=[AAAAAAAAAAAAAAAAAAAAAA, AAAAAAAAAAAAAAAAAAAAAA, AAAAAAAAAAAAAAAAAAAAAA], " +
"isr=[2, 3], removingReplicas=[], addingReplicas=[], elr=[], lastKnownElr=[], leader=2, " +
"leaderRecoveryState=RECOVERED, leaderEpoch=1, partitionEpoch=345)", stringifier.toString());
"leaderRecoveryState=RECOVERED, leaderEpoch=1, partitionEpoch=345, " +
"classicToDisklessStartOffset=-1, disklessProducerStates=[], disklessLeaderEpoch=-1)", stringifier.toString());
}

@Test
Expand Down
2 changes: 2 additions & 0 deletions .../org/apache/kafka/streams/processor/internals/assignment/TaskAssignorConvergenceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.test.api.Flaky;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.StreamsConfig;
Expand Down Expand Up @@ -475,6 +476,7 @@ public void droppingNodesShouldConverge(final String rackAwareStrategy) {
StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_MIN_TRAFFIC,
StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_BALANCE_SUBTOPOLOGY
})
@Flaky(value = "KAFKA-16491", comment = "Flaky for rackAwareStrategy=balance_subtopology.")
public void randomClusterPerturbationsShouldConverge(final String rackAwareStrategy) {
setUp(rackAwareStrategy);
// do as many tests as we can in 10 seconds
Expand Down
Loading