import java.util.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
-import java.util.regex.Pattern;
import java.util.stream.Collectors;
@Slf4j
public class SplitterStreamProcessor implements Runnable
{
- final static Pattern PATTERN = Pattern.compile("\\W+");
-
+ private final MessageSplitter splitter;
private final String inputTopic;
private final String outputTopic;
private final KafkaConsumer<String, String> consumer;
private long lastCommit;
public SplitterStreamProcessor(
+ MessageSplitter splitter,
SplitterApplicationProperties properties,
Clock clock,
TaskExecutor executor)
{
+ this.splitter = splitter;
+
this.inputTopic = properties.getInputTopic();
this.outputTopic = properties.getOutputTopic();
offsets[inputRecord.partition()] = inputRecord.offset();
leaderEpochs[inputRecord.partition()] = inputRecord.leaderEpoch();
- String[] words = PATTERN.split(inputRecord.value());
+ String[] words = splitter.split(inputRecord.value());
for (int i = 0; i < words.length; i++)
{
ProducerRecord<String, String> outputRecord =