Skip to content

Commit 9f1183c

Browse files
author
h1alexbel
committed
1 parent 5e71c5e commit 9f1183c

4 files changed

Lines changed: 217 additions & 21 deletions

File tree

‎src/main/java/io/github/eocqrs/kafka/fake/FkConsumer.java‎

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -82,14 +82,21 @@ public void subscribe(final Collection<String> topics) {
8282
);
8383
}
8484

85-
/*
86-
* @todo #54:60m/DEV Fake subscribe with
87-
* ConsumerRebalanceListener is not implemented
88-
*/
8985
@Override
9086
public void subscribe(final ConsumerRebalanceListener listener,
9187
final String... topics) {
92-
throw new UnsupportedOperationException("#subscribe()");
88+
new ListOf<>(topics)
89+
.forEach(t -> {
90+
try {
91+
this.broker.with(
92+
new WithRebalanceListener(
93+
new SubscribeDirs(t, this.id),
94+
listener
95+
).value());
96+
} catch (final Exception ex) {
97+
throw new IllegalStateException(ex);
98+
}
99+
});
93100
}
94101

95102
/*

‎src/main/java/io/github/eocqrs/kafka/fake/WithRebalanceListener.java‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
/*
2+
* The MIT License (MIT)
3+
*
4+
* Copyright (c) 2023 Aliaksei Bialiauski, EO-CQRS
5+
*
6+
* Permission is hereby granted, free of charge, to any person obtaining a copy
7+
* of this software and associated documentation files (the "Software"), to deal
8+
* in the Software without restriction, including without limitation the rights
9+
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
10+
* copies of the Software, and to permit persons to whom the Software is
11+
* furnished to do so, subject to the following conditions:
12+
*
13+
* The above copyright notice and this permission notice shall be included
14+
* in all copies or substantial portions of the Software.
15+
*
16+
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17+
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18+
* FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
19+
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20+
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21+
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
22+
* SOFTWARE.
23+
*/
24+
25+
package io.github.eocqrs.kafka.fake;
26+
27+
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
28+
import org.cactoos.Scalar;
29+
import org.xembly.Directives;
30+
31+
/**
32+
* Directives with Rebalance Listener.
33+
*
34+
* @author Aliaksei Bialiauski (abialiauski.dev@gmail.com)
35+
* @see org.apache.kafka.clients.consumer.ConsumerRebalanceListener
36+
* @since 0.3.5
37+
*/
38+
public final class WithRebalanceListener implements Scalar<Directives> {
39+
40+
/**
41+
* Origin Directives.
42+
*/
43+
private final Scalar<Directives> dirs;
44+
/**
45+
* ConsumerRebalanceListener.
46+
*/
47+
private final ConsumerRebalanceListener listener;
48+
49+
/**
50+
* Ctor.
51+
*
52+
* @param drs Directives
53+
* @param lstnr Listener
54+
*/
55+
public WithRebalanceListener(
56+
final Scalar<Directives> drs,
57+
final ConsumerRebalanceListener lstnr
58+
) {
59+
this.dirs = drs;
60+
this.listener = lstnr;
61+
}
62+
63+
@Override
64+
public Directives value() throws Exception {
65+
return this.dirs
66+
.value()
67+
.up()
68+
.addIf("listener")
69+
.set(this.listener.toString());
70+
}
71+
}

‎src/test/java/io/github/eocqrs/kafka/fake/FkConsumerTest.java‎

Lines changed: 58 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -25,9 +25,7 @@
2525
package io.github.eocqrs.kafka.fake;
2626

2727
import com.jcabi.log.Logger;
28-
import com.jcabi.xml.XML;
2928
import io.github.eocqrs.kafka.Consumer;
30-
import io.github.eocqrs.xfake.FkStorage;
3129
import io.github.eocqrs.xfake.InFile;
3230
import io.github.eocqrs.xfake.Synchronized;
3331
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
@@ -37,15 +35,13 @@
3735
import org.hamcrest.Matchers;
3836
import org.junit.jupiter.api.Assertions;
3937
import org.junit.jupiter.api.BeforeEach;
40-
import org.junit.jupiter.api.Disabled;
4138
import org.junit.jupiter.api.Test;
4239
import org.junit.jupiter.api.extension.ExtendWith;
4340
import org.mockito.Mock;
4441
import org.mockito.junit.jupiter.MockitoExtension;
4542

4643
import java.io.IOException;
4744
import java.time.Duration;
48-
import java.util.ArrayList;
4945
import java.util.Collection;
5046
import java.util.UUID;
5147
import java.util.logging.Level;
@@ -139,6 +135,64 @@ void subscribesToTopics() throws Exception {
139135
consumer.close();
140136
}
141137

138+
@Test
139+
void subscribesWithRebalanceListener() throws Exception {
140+
final String topic = "listener.test";
141+
final String rebalance = "rebalance";
142+
final UUID uuid = UUID.fromString("e343a512-9d02-40e8-92b6-1538014d3975");
143+
final FkBroker with = this.broker
144+
.with(new TopicDirs(topic).value());
145+
final Consumer<String, String> consumer =
146+
new FkConsumer<>(
147+
uuid,
148+
with
149+
);
150+
final ConsumerRebalanceListener listener = new ConsumerRebalanceListener() {
151+
@Override
152+
public void onPartitionsRevoked(final Collection<TopicPartition> collection) {
153+
}
154+
155+
@Override
156+
public void onPartitionsAssigned(final Collection<TopicPartition> collection) {
157+
}
158+
159+
@Override
160+
public String toString() {
161+
return rebalance;
162+
}
163+
};
164+
consumer.subscribe(listener, topic);
165+
MatcherAssert.assertThat(
166+
"topic subscriptions in right format",
167+
with.data(
168+
"broker/subs/sub[topic = '%s']/topic/text()"
169+
.formatted(
170+
topic
171+
)
172+
),
173+
Matchers.contains(topic)
174+
);
175+
MatcherAssert.assertThat(
176+
"Consumer ID in right format",
177+
with.data(
178+
"broker/subs/sub[consumer = '%s']/consumer/text()"
179+
.formatted(
180+
uuid.toString()
181+
)
182+
),
183+
Matchers.contains(uuid.toString())
184+
);
185+
MatcherAssert.assertThat(
186+
"Consumer Rebalance Listener in right format",
187+
with.data(
188+
"broker/subs/sub[listener = '%s']/listener/text()"
189+
.formatted(listener.toString())
190+
),
191+
Matchers.contains(rebalance)
192+
);
193+
consumer.close();
194+
}
195+
142196
@Test
143197
void subscribesWithVarArgs() throws Exception {
144198
final String topic = "varargs.test";
@@ -171,18 +225,6 @@ void createsFakeConsumer() {
171225
final FkConsumer<String, String> consumer =
172226
new FkConsumer<>(UUID.randomUUID(), this.broker);
173227
MatcherAssert.assertThat(consumer, Matchers.is(Matchers.notNullValue()));
174-
assertThrows(
175-
UnsupportedOperationException.class,
176-
() -> consumer.subscribe(new ConsumerRebalanceListener() {
177-
@Override
178-
public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {
179-
}
180-
181-
@Override
182-
public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {
183-
}
184-
}, "fake")
185-
);
186228
assertThrows(
187229
UnsupportedOperationException.class,
188230
() -> consumer.records("123", Duration.ofMillis(100L))

‎src/test/java/io/github/eocqrs/kafka/fake/WithRebalanceListenerTest.java‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
/*
2+
* The MIT License (MIT)
3+
*
4+
* Copyright (c) 2023 Aliaksei Bialiauski, EO-CQRS
5+
*
6+
* Permission is hereby granted, free of charge, to any person obtaining a copy
7+
* of this software and associated documentation files (the "Software"), to deal
8+
* in the Software without restriction, including without limitation the rights
9+
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
10+
* copies of the Software, and to permit persons to whom the Software is
11+
* furnished to do so, subject to the following conditions:
12+
*
13+
* The above copyright notice and this permission notice shall be included
14+
* in all copies or substantial portions of the Software.
15+
*
16+
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17+
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18+
* FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
19+
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20+
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21+
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
22+
* SOFTWARE.
23+
*/
24+
25+
package io.github.eocqrs.kafka.fake;
26+
27+
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
28+
import org.apache.kafka.common.TopicPartition;
29+
import org.hamcrest.MatcherAssert;
30+
import org.hamcrest.Matchers;
31+
import org.junit.jupiter.api.Test;
32+
33+
import java.util.Collection;
34+
import java.util.UUID;
35+
36+
/**
37+
* Test case for {@link WithRebalanceListener}.
38+
*
39+
* @author Aliaksei Bialiauski (abialiauski.dev@gmail.com)
40+
* @since 0.3.5
41+
*/
42+
final class WithRebalanceListenerTest {
43+
44+
@Test
45+
void dirsInRightFormat() throws Exception {
46+
final UUID id = UUID.fromString("a6d0094c-d393-4d63-bc4c-20b33b38bc8f");
47+
final String rebalance = "rebalance";
48+
final String directives =
49+
"XPATH \"broker/subs\";ADD \"sub\";ADDIF \"topic\";SET \"test\";UP;ADDIF "
50+
+ "\"consumer\";SET \"a6d0094c-d393-4d63-bc4c-20b33b38bc8f\";"
51+
+ "\n7:UP;ADDIF \"listener\";SET \"rebalance\";";
52+
MatcherAssert.assertThat(
53+
"Directives in the right format",
54+
new WithRebalanceListener(new SubscribeDirs("test", id),
55+
new ConsumerRebalanceListener() {
56+
@Override
57+
public void onPartitionsRevoked(final Collection<TopicPartition> collection) {
58+
}
59+
60+
@Override
61+
public void onPartitionsAssigned(final Collection<TopicPartition> collection) {
62+
}
63+
64+
@Override
65+
public String toString() {
66+
return rebalance;
67+
}
68+
})
69+
.value()
70+
.toString(),
71+
Matchers.equalTo(
72+
directives
73+
)
74+
);
75+
}
76+
}

0 commit comments

Comments
 (0)