* Hier wurde die Deaktivierung der Commits im Falle eines Fachlogik-Fehlers
zurückgebaut, um die einzelnen Schritte der Übung leichter
nachvollziehbar zu machen.
* Die Deaktivierung der Commits im Falle eines Fachlogik-Fehlers (aka
Exception in der Fachlogik) ist nötig, da ansonsten in dem Rebalance,
das durch das `unsubscribe()` ausgelöst wird, die aktuelle
Offset-Position aus sicht der Kafka-Client-Library commited wird.
* *Dies führt dazu, dass der zugehörige Testfall fehl schlägt*
* Dies würde dann zu einem Verlust von Nachrichten führen, da die
Client-Bibliothek davon ausgeht, dass alle vom letzten `poll()`
gelieferten Nachrichten erfolgreich verarbeitet wurden.
* D.h., nach einem Fachlogik-Fehler darf _kein_ Commit über die
Convenient-Methode des `KafkaConsumer` durchgeführt werden, die
automatisch die Offsets bestätigt, die die Bibliothek für aktuell
hält.
]]
then
docker-compose rm -svf adder-1 adder-2
]]
then
docker-compose rm -svf adder-1 adder-2
- mvn clean install || exit
+ mvn -D skipTests clean install || exit
else
echo "Using image existing images:"
docker image ls $IMAGE
else
echo "Using image existing images:"
docker image ls $IMAGE
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
stateRepository,
properties.getClientId(),
properties.getTopic(),
stateRepository,
properties.getClientId(),
properties.getTopic(),
- Clock.systemDefaultZone(),
- properties.getCommitInterval(),
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.TopicPartition;
-import java.time.Clock;
-import java.time.Duration;
-import java.time.Instant;
private final StateRepository stateRepository;
private final String id;
private final String topic;
private final StateRepository stateRepository;
private final String id;
private final String topic;
- private final Clock clock;
- private final Duration commitInterval;
private final Consumer consumer;
private final Set<Integer> partitions = new HashSet<>();
private final Consumer consumer;
private final Set<Integer> partitions = new HashSet<>();
- private Instant lastCommit = Instant.EPOCH;
- private boolean commitsEnabled = true;
-
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions)
{
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions)
{
Integer partition = tp.partition();
log.info("{} - removing partition: {}", id, partition);
this.partitions.remove(partition);
Integer partition = tp.partition();
log.info("{} - removing partition: {}", id, partition);
this.partitions.remove(partition);
+ Map<String, AdderResult> state = recordHandler.removePartition(partition);
+ Long offset = consumer.position(tp);
+ log.info(
+ "{} - offset of next unseen message for partition {} is {}",
+ id,
+ partition,
+ offset);
+ for (String user : state.keySet())
- Map<String, AdderResult> state = recordHandler.removePartition(partition);
- Long offset = consumer.position(tp);
- "{} - offset of next unseen message for partition {} is {}",
+ "{} - Saved state for partition={}|user={}: {}",
- offset);
- for (String user : state.keySet())
- {
- log.info(
- "{} - Saved state for partition={}|user={}: {}",
- id,
- partition,
- user,
- state.get(user));
- }
- Map<String, List<AdderResult>> results = adderResults.removePartition(partition);
- stateRepository.save(new StateDocument(partition, state, results, offset));
- }
- else
- {
- log.info("{} - Offset commits are disabled! Last commit: {}", id, lastCommit);
+ user,
+ state.get(user));
+ Map<String, List<AdderResult>> results = adderResults.removePartition(partition);
+ stateRepository.save(new StateDocument(partition, state, results, offset));
@Override
public void beforeNextPoll()
{
@Override
public void beforeNextPoll()
{
- if (!commitsEnabled)
- {
- log.info("{} - Offset commits are disabled! Last commit: {}", id, lastCommit);
- return;
- }
-
- if (lastCommit.plus(commitInterval).isBefore(clock.instant()))
- {
- partitions
- .stream()
- .forEach(partition ->
+ partitions
+ .stream()
+ .forEach(partition ->
+ {
+ log.info("{} - persisting state & offset for partition: {}", id, partition);
+ Map<String, AdderResult> state = recordHandler.getState(partition).getState();
+ Long offset = consumer.position(new TopicPartition(topic, partition));
+ log.info(
+ "{} - offset of next unseen message for partition {} is {}",
+ id,
+ partition,
+ offset);
+ for (String user : state.keySet())
- log.info("{} - persisting state & offset for partition: {}", id, partition);
- Map<String, AdderResult> state = recordHandler.getState(partition).getState();
- Long offset = consumer.position(new TopicPartition(topic, partition));
- "{} - offset of next unseen message for partition {} is {}",
+ "{} - Saved state for partition={}|user={}: {}",
- offset);
- for (String user : state.keySet())
- {
- log.info(
- "{} - Saved state for partition={}|user={}: {}",
- id,
- partition,
- user,
- state.get(user));
- }
- Map<String, List<AdderResult>> results = adderResults.getState(partition);
- stateRepository.save(new StateDocument(partition, state, results, offset));
- });
-
- lastCommit = clock.instant();
- }
- }
-
- @Override
- public void enableCommits()
- {
- commitsEnabled = true;
- }
-
- @Override
- public void disableCommits()
- {
- commitsEnabled = false;
+ user,
+ state.get(user));
+ }
+ Map<String, List<AdderResult>> results = adderResults.getState(partition);
+ stateRepository.save(new StateDocument(partition, state, results, offset));
+ });
try
{
log.info("{} - Subscribing to topic {}", id, topic);
try
{
log.info("{} - Subscribing to topic {}", id, topic);
- rebalanceListener.enableCommits();
consumer.subscribe(Arrays.asList(topic), rebalanceListener);
while (true)
consumer.subscribe(Arrays.asList(topic), rebalanceListener);
while (true)
- log.error("{} - Unexpected error: {}, disabling commits", id, e.toString(), e);
- rebalanceListener.disableCommits();
+ log.error("{} - Unexpected error: {}", id, e.toString(), e);
public interface RebalanceListener extends ConsumerRebalanceListener
{
default void beforeNextPoll() {}
public interface RebalanceListener extends ConsumerRebalanceListener
{
default void beforeNextPoll() {}
-
- default void enableCommits() {}
- default void disableCommits() {}