Skip to content

Fix Kafka Connect data streams test flake #12187

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
AlexeyKuznetsov-DD wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Fix Kafka Connect data streams test flake #12187

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
66 changes: 39 additions & 27 deletions ...entation/kafka/kafka-connect-0.11/src/test/groovy/ConnectWorkerInstrumentationTest.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,11 @@ import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit

class ConnectWorkerInstrumentationTest extends InstrumentationSpecification {
private static final String SOURCE_TOPIC = 'source-test-topic'
private static final String SINK_TOPIC = 'sink-test-topic'

@Shared
EmbeddedKafkaBroker embeddedKafka = new EmbeddedKafkaBroker(1, false, 1, 'test-topic')
EmbeddedKafkaBroker embeddedKafka = new EmbeddedKafkaBroker(1, false, 1, SOURCE_TOPIC, SINK_TOPIC)

def setupSpec() {
embeddedKafka.afterPropertiesSet() // Initializes the broker
Expand Down Expand Up @@ -97,7 +100,7 @@ class ConnectWorkerInstrumentationTest extends InstrumentationSpecification {
'connector.class': 'org.apache.kafka.connect.file.FileStreamSourceConnector',
'tasks.max' : '1',
'file' : tempFile.getAbsolutePath(),
'topic' : 'test-topic'
'topic' : SOURCE_TOPIC
]

// Latch to wait for connector addition
Expand Down Expand Up @@ -133,7 +136,7 @@ class ConnectWorkerInstrumentationTest extends InstrumentationSpecification {
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps)
consumer.subscribe(['test-topic'])
consumer.subscribe([SOURCE_TOPIC])

String receivedMessage = null
for (int i = 0; i < 10; i++) {
Expand All @@ -144,31 +147,25 @@ class ConnectWorkerInstrumentationTest extends InstrumentationSpecification {
break
}
}
TEST_DATA_STREAMS_WRITER.waitForGroups(2)

then:
receivedMessage == "Hello Kafka"

StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
it.parentHash == 0
}
List<StatsGroup> pathway = waitForPathway(SOURCE_TOPIC, "test-consumer-group")
StatsGroup first = pathway[0]
verifyAll(first) {
tags.hasAllTags(
"direction:out",
"topic:test-topic",
"topic:" + SOURCE_TOPIC,
"type:kafka"
)
}

StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find {
it.parentHash == first.hash
}
StatsGroup second = pathway[1]
verifyAll(second) {
tags.hasAllTags("direction:in", "group:test-consumer-group", "topic:test-topic", "type:kafka")
tags.hasAllTags("direction:in", "group:test-consumer-group", "topic:" + SOURCE_TOPIC, "type:kafka")
}
TEST_DATA_STREAMS_WRITER.getServices().contains('file-source-connector')


cleanup:
consumer?.close()
herder?.stop()
Expand Down Expand Up @@ -233,7 +230,7 @@ class ConnectWorkerInstrumentationTest extends InstrumentationSpecification {
'connector.class': 'org.apache.kafka.connect.file.FileStreamSinkConnector',
'tasks.max' : '1',
'file' : sinkFile.getAbsolutePath(),
'topics' : 'test-topic'
'topics' : SINK_TOPIC
]

// Latch to wait for connector addition
Expand Down Expand Up @@ -265,7 +262,7 @@ class ConnectWorkerInstrumentationTest extends InstrumentationSpecification {
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps)
producer.send(new ProducerRecord<>("test-topic", "key1", "Hello Kafka Sink"))
producer.send(new ProducerRecord<>(SINK_TOPIC, "key1", "Hello Kafka Sink"))
producer.flush()
producer.close()

Expand All @@ -278,33 +275,48 @@ class ConnectWorkerInstrumentationTest extends InstrumentationSpecification {
}

String fileContents = sinkFile.text
TEST_DATA_STREAMS_WRITER.waitForGroups(2)

then:
fileContents.contains("Hello Kafka Sink")

StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
it.parentHash == 0
}
List<StatsGroup> pathway = waitForPathway(SINK_TOPIC, "connect-file-sink-connector")
StatsGroup first = pathway[0]
verifyAll(first) {
tags.hasAllTags("direction:out", "topic:test-topic", "type:kafka")
tags.hasAllTags("direction:out", "topic:" + SINK_TOPIC, "type:kafka")
}

StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find {
it.parentHash == first.hash
}
StatsGroup second = pathway[1]
verifyAll(second) {
tags.hasAllTags("direction:in", "group:connect-file-sink-connector", "topic:test-topic", "type:kafka")
tags.hasAllTags("direction:in", "group:connect-file-sink-connector", "topic:" + SINK_TOPIC, "type:kafka")
}
TEST_DATA_STREAMS_WRITER.getServices().contains('file-sink-connector')


cleanup:
herder?.stop()
worker?.stop()
sinkFile?.delete()
}

private List<StatsGroup> waitForPathway(String topic, String group) {
long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(10)
while (System.currentTimeMillis() < deadline) {
StatsGroup producer = TEST_DATA_STREAMS_WRITER.groups.find {
it.parentHash == 0 &&
it.tags.hasAllTags("direction:out", "topic:" + topic, "type:kafka")
}
if (producer != null) {
StatsGroup consumer = TEST_DATA_STREAMS_WRITER.groups.find {
it.parentHash == producer.hash &&
it.tags.hasAllTags("direction:in", "group:" + group, "topic:" + topic, "type:kafka")
}
if (consumer != null) {
return [producer, consumer]
}
}
Thread.sleep(20)
}
throw new AssertionError("No matching data-stream pathway found for topic " + topic + " and group " + group)
}

@Override
protected boolean isDataStreamsEnabled() {
return true
Expand Down
Loading