test fixes
This commit is contained in:
parent
cbc3f174b5
commit
eade9d86df
1 changed files with 9 additions and 3 deletions
|
@ -85,9 +85,15 @@ class ConsumerOffsetsSerdeTest extends AbstractIntegrationTest {
|
||||||
.atMost(Duration.ofMinutes(1))
|
.atMost(Duration.ofMinutes(1))
|
||||||
.untilAsserted(() -> {
|
.untilAsserted(() -> {
|
||||||
for (var rec : consumer.poll(Duration.ofMillis(200))) {
|
for (var rec : consumer.poll(Duration.ofMillis(200))) {
|
||||||
DeserializeResult key = keyDeserializer.deserialize(null, rec.key().get());
|
DeserializeResult key = rec.key() != null
|
||||||
DeserializeResult val = valueDeserializer.deserialize(null, rec.value().get());
|
? keyDeserializer.deserialize(null, rec.key().get())
|
||||||
polled.add(Tuples.of(key, val));
|
: null;
|
||||||
|
DeserializeResult val = rec.value() != null
|
||||||
|
? valueDeserializer.deserialize(null, rec.value().get())
|
||||||
|
: null;
|
||||||
|
if (key != null && val != null) {
|
||||||
|
polled.add(Tuples.of(key, val));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
assertThat(polled).anyMatch(t -> isCommitMessage(t.getT1(), t.getT2()));
|
assertThat(polled).anyMatch(t -> isCommitMessage(t.getT1(), t.getT2()));
|
||||||
assertThat(polled).anyMatch(t -> isGroupMetadataMessage(t.getT1(), t.getT2()));
|
assertThat(polled).anyMatch(t -> isGroupMetadataMessage(t.getT1(), t.getT2()));
|
||||||
|
|
Loading…
Add table
Reference in a new issue