projects
/
demos
/
kafka
/
wordcount
/ commitdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
| commitdiff |
tree
raw
|
patch
|
inline
| side by side (parent:
b67e6b9
)
counter: 1.2.12 - Renamed `WordCount` into `WordCounter` -- ALIGN
counter-1.2.12
author
Kai Moritz
<kai@juplo.de>
Tue, 14 May 2024 19:54:28 +0000
(21:54 +0200)
committer
Kai Moritz
<kai@juplo.de>
Tue, 14 May 2024 20:37:41 +0000
(22:37 +0200)
src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java
patch
|
blob
|
history
src/main/java/de/juplo/kafka/wordcount/counter/WordCounter.java
patch
|
blob
|
history
src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java
patch
|
blob
|
history
src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
patch
|
blob
|
history
src/test/java/de/juplo/kafka/wordcount/counter/TestData.java
patch
|
blob
|
history
diff --git
a/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java
b/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java
index
3dd8c0f
..
b1343a7
100644
(file)
--- a/
src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java
+++ b/
src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java
@@
-50,7
+50,7
@@
public class CounterStreamProcessor
.groupByKey()
.count(Materialized.as(storeSupplier))
.toStream()
.groupByKey()
.count(Materialized.as(storeSupplier))
.toStream()
- .map((word, count
) -> new KeyValue<>(word, WordCount.of(word, count
)))
+ .map((word, count
er) -> new KeyValue<>(word, WordCounter.of(word, counter
)))
.to(outputTopic);
Topology topology = builder.build();
.to(outputTopic);
Topology topology = builder.build();
diff --git
a/src/main/java/de/juplo/kafka/wordcount/counter/WordCounter.java
b/src/main/java/de/juplo/kafka/wordcount/counter/WordCounter.java
index
958f9b7
..
1334e5b
100644
(file)
--- a/
src/main/java/de/juplo/kafka/wordcount/counter/WordCounter.java
+++ b/
src/main/java/de/juplo/kafka/wordcount/counter/WordCounter.java
@@
-8,14
+8,14
@@
import lombok.NoArgsConstructor;
@Data
@NoArgsConstructor
@AllArgsConstructor(staticName = "of")
@Data
@NoArgsConstructor
@AllArgsConstructor(staticName = "of")
-public class WordCount
+public class WordCount
er
{
String user;
String word;
{
String user;
String word;
- long count;
+ long count
er
;
- public static WordCount
of(Word word, long count
)
+ public static WordCount
er of(Word word, long counter
)
{
{
- return new WordCount
(word.getUser(), word.getWord(), count
);
+ return new WordCount
er(word.getUser(), word.getWord(), counter
);
}
}
}
}
diff --git
a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java
b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java
index
e0f4672
..
5a3507a
100644
(file)
--- a/
src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java
+++ b/
src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java
@@
-76,16
+76,16
@@
public class CounterApplicationIT
@RequiredArgsConstructor
static class Consumer
{
@RequiredArgsConstructor
static class Consumer
{
- private final List<KeyValue<Word, WordCount>> received = new LinkedList<>();
+ private final List<KeyValue<Word, WordCount
er
>> received = new LinkedList<>();
@KafkaListener(groupId = "TEST", topics = TOPIC_OUT)
@KafkaListener(groupId = "TEST", topics = TOPIC_OUT)
- public synchronized void receive(ConsumerRecord<Word, WordCount> record)
+ public synchronized void receive(ConsumerRecord<Word, WordCount
er
> record)
{
log.debug("Received message: {}", record);
received.add(KeyValue.pair(record.key(),record.value()));
}
{
log.debug("Received message: {}", record);
received.add(KeyValue.pair(record.key(),record.value()));
}
- synchronized List<KeyValue<Word, WordCount>> getReceivedMessages()
+ synchronized List<KeyValue<Word, WordCount
er
>> getReceivedMessages()
{
return received;
}
{
return received;
}
diff --git
a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
index
159123b
..
b785dfa
100644
(file)
--- a/
src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
+++ b/
src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
@@
-45,14
+45,14
@@
public class CounterStreamProcessorTopologyTest
(JsonSerializer<String>)keySerde.serializer(),
(JsonSerializer<Word>)valueSerde.serializer());
(JsonSerializer<String>)keySerde.serializer(),
(JsonSerializer<Word>)valueSerde.serializer());
- TestOutputTopic<Word, WordCount> out = testDriver.createOutputTopic(
+ TestOutputTopic<Word, WordCount
er
> out = testDriver.createOutputTopic(
OUT,
(JsonDeserializer<Word>)keySerde.deserializer(),
OUT,
(JsonDeserializer<Word>)keySerde.deserializer(),
- (JsonDeserializer<WordCount>)valueSerde.deserializer());
+ (JsonDeserializer<WordCount
er
>)valueSerde.deserializer());
TestData.writeInputData((key, value) -> in.pipeInput(key, value));
TestData.writeInputData((key, value) -> in.pipeInput(key, value));
- List<KeyValue<Word,
WordCount
>> receivedMessages = out
+ List<KeyValue<Word,
WordCounter
>> receivedMessages = out
.readRecordsToList()
.stream()
.map(record -> KeyValue.pair(record.key(), record.value()))
.readRecordsToList()
.stream()
.map(record -> KeyValue.pair(record.key(), record.value()))
diff --git
a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java
b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java
index
c1dd45a
..
43e1919
100644
(file)
--- a/
src/test/java/de/juplo/kafka/wordcount/counter/TestData.java
+++ b/
src/test/java/de/juplo/kafka/wordcount/counter/TestData.java
@@
-50,7
+50,7
@@
class TestData
Word.of("klaus","s"));
}
Word.of("klaus","s"));
}
- static void assertExpectedResult(List<KeyValue<Word, WordCount>> receivedMessages)
+ static void assertExpectedResult(List<KeyValue<Word, WordCount
er
>> receivedMessages)
{
assertThat(receivedMessages).hasSize(11);
assertThat(receivedMessages).containsSubsequence(
{
assertThat(receivedMessages).hasSize(11);
assertThat(receivedMessages).containsSubsequence(
@@
-71,41
+71,41
@@
class TestData
expectedMessages[9]); // Boäh
}
expectedMessages[9]); // Boäh
}
- static KeyValue<Word,
WordCount
>[] expectedMessages = new KeyValue[]
+ static KeyValue<Word,
WordCounter
>[] expectedMessages = new KeyValue[]
{
KeyValue.pair(
Word.of("peter","Hallo"),
{
KeyValue.pair(
Word.of("peter","Hallo"),
- WordCount.of("peter","Hallo",1)),
+ WordCount
er
.of("peter","Hallo",1)),
KeyValue.pair(
Word.of("klaus","Müsch"),
KeyValue.pair(
Word.of("klaus","Müsch"),
- WordCount.of("klaus","Müsch",1)),
+ WordCount
er
.of("klaus","Müsch",1)),
KeyValue.pair(
Word.of("peter","Welt"),
KeyValue.pair(
Word.of("peter","Welt"),
- WordCount.of("peter","Welt",1)),
+ WordCount
er
.of("peter","Welt",1)),
KeyValue.pair(
Word.of("klaus","Müsch"),
KeyValue.pair(
Word.of("klaus","Müsch"),
- WordCount.of("klaus","Müsch",2)),
+ WordCount
er
.of("klaus","Müsch",2)),
KeyValue.pair(
Word.of("klaus","s"),
KeyValue.pair(
Word.of("klaus","s"),
- WordCount.of("klaus","s",1)),
+ WordCount
er
.of("klaus","s",1)),
KeyValue.pair(
Word.of("peter","Boäh"),
KeyValue.pair(
Word.of("peter","Boäh"),
- WordCount.of("peter","Boäh",1)),
+ WordCount
er
.of("peter","Boäh",1)),
KeyValue.pair(
Word.of("peter","Welt"),
KeyValue.pair(
Word.of("peter","Welt"),
- WordCount.of("peter","Welt",2)),
+ WordCount
er
.of("peter","Welt",2)),
KeyValue.pair(
Word.of("peter","Boäh"),
KeyValue.pair(
Word.of("peter","Boäh"),
- WordCount.of("peter","Boäh",2)),
+ WordCount
er
.of("peter","Boäh",2)),
KeyValue.pair(
Word.of("klaus","s"),
KeyValue.pair(
Word.of("klaus","s"),
- WordCount.of("klaus","s",2)),
+ WordCount
er
.of("klaus","s",2)),
KeyValue.pair(
Word.of("peter","Boäh"),
KeyValue.pair(
Word.of("peter","Boäh"),
- WordCount.of("peter","Boäh",3)),
+ WordCount
er
.of("peter","Boäh",3)),
KeyValue.pair(
Word.of("klaus","s"),
KeyValue.pair(
Word.of("klaus","s"),
- WordCount.of("klaus","s",3)),
+ WordCount
er
.of("klaus","s",3)),
};
static Map<String, Object> convertToMap(Properties properties)
};
static Map<String, Object> convertToMap(Properties properties)