Skip to content

Commit 0a444ab

Browse files
jolarsenjolarsen
authored
Felle 708. Nytt kafkaoppsett (#2546)
1 parent 23beefb commit 0a444ab

File tree

6 files changed

+37
-73
lines changed

6 files changed

+37
-73
lines changed

‎domenetjenester/lonnskomp/pom.xml‎

Lines changed: 0 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -38,18 +38,6 @@
3838
<groupId>no.nav.foreldrepenger.abakus</groupId>
3939
<artifactId>vedtak</artifactId>
4040
</dependency>
41-
<dependency>
42-
<groupId>io.swagger.core.v3</groupId>
43-
<artifactId>swagger-annotations-jakarta</artifactId>
44-
</dependency>
45-
<dependency>
46-
<groupId>org.apache.kafka</groupId>
47-
<artifactId>kafka-streams</artifactId>
48-
</dependency>
49-
<dependency>
50-
<groupId>no.nav.foreldrepenger.abakus</groupId>
51-
<artifactId>abakus-kontrakt</artifactId>
52-
</dependency>
5341
<dependency>
5442
<groupId>no.nav.foreldrepenger.abakus</groupId>
5543
<artifactId>felles</artifactId>

‎domenetjenester/vedtak/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@
5151
</dependency>
5252
<dependency>
5353
<groupId>org.apache.kafka</groupId>
54-
<artifactId>kafka-streams</artifactId>
54+
<artifactId>kafka-clients</artifactId>
5555
</dependency>
5656

5757
<dependency>

‎domenetjenester/vedtak/src/main/java/no/nav/foreldrepenger/abakus/vedtak/kafka/VedtakConsumer.java‎

Lines changed: 12 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,13 @@
11
package no.nav.foreldrepenger.abakus.vedtak.kafka;
22

3-
import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
3+
import java.util.List;
44

5-
import java.time.Duration;
6-
import java.time.temporal.ChronoUnit;
7-
8-
import jakarta.enterprise.context.ApplicationScoped;
9-
import jakarta.inject.Inject;
10-
11-
import org.apache.kafka.streams.KafkaStreams;
12-
import org.apache.kafka.streams.StreamsBuilder;
13-
import org.apache.kafka.streams.Topology;
14-
import org.apache.kafka.streams.kstream.Consumed;
155
import org.slf4j.Logger;
166
import org.slf4j.LoggerFactory;
177

18-
import no.nav.foreldrepenger.konfig.KonfigVerdi;
19-
import no.nav.vedtak.felles.integrasjon.kafka.KafkaProperties;
8+
import jakarta.enterprise.context.ApplicationScoped;
9+
import jakarta.inject.Inject;
10+
import no.nav.vedtak.felles.integrasjon.kafka.KafkaConsumerManager;
2011
import no.nav.vedtak.log.metrics.Controllable;
2112
import no.nav.vedtak.log.metrics.LiveAndReadinessAware;
2213

@@ -25,47 +16,20 @@ public class VedtakConsumer implements LiveAndReadinessAware, Controllable {
2516

2617
private static final Logger LOG = LoggerFactory.getLogger(VedtakConsumer.class);
2718

28-
private static final String APPLICATION_ID = "fpabakus"; // Hold konstant pga offset commit
29-
30-
private KafkaStreams stream;
19+
private KafkaConsumerManager<String, String> kcm;
3120
private String topic;
3221

3322
VedtakConsumer() {
3423
}
3524

3625
@Inject
37-
public VedtakConsumer(@KonfigVerdi(value = "kafka.fattevedtak.topic", defaultVerdi = "teamforeldrepenger.familie-vedtakfattet-v1") String topicName,
38-
VedtaksHendelseHåndterer vedtaksHendelseHåndterer) {
39-
this.topic = topicName;
40-
41-
final Consumed<String, String> consumed = Consumed.with(Topology.AutoOffsetReset.EARLIEST);
42-
43-
final var builder = new StreamsBuilder();
44-
builder.stream(topic, consumed).foreach(vedtaksHendelseHåndterer::handleMessage);
45-
46-
this.stream = new KafkaStreams(builder.build(), KafkaProperties.forStreamsStringValue(APPLICATION_ID));
26+
public VedtakConsumer(VedtaksHendelseHåndterer vedtaksHendelseHåndterer) {
27+
this.kcm = new KafkaConsumerManager<>(List.of(vedtaksHendelseHåndterer));
4728
}
4829

49-
private void addShutdownHooks() {
50-
stream.setStateListener((newState, oldState) -> {
51-
LOG.info("{} :: From state={} to state={}", topic, oldState, newState);
52-
53-
if (newState == KafkaStreams.State.ERROR) {
54-
// if the stream has died there is no reason to keep spinning
55-
LOG.warn("{} :: No reason to keep living, closing stream", topic);
56-
stop();
57-
}
58-
});
59-
stream.setUncaughtExceptionHandler(ex -> {
60-
LOG.error(topic + " :: Caught exception in stream, exiting", ex);
61-
return SHUTDOWN_CLIENT;
62-
});
63-
}
64-
65-
6630
@Override
6731
public boolean isAlive() {
68-
return stream != null && stream.state().isRunningOrRebalancing();
32+
return kcm.allRunning();
6933
}
7034

7135
@Override
@@ -75,15 +39,13 @@ public boolean isReady() {
7539

7640
@Override
7741
public void start() {
78-
addShutdownHooks();
79-
stream.start();
80-
LOG.info("Starter konsumering av topic={}, tilstand={}", topic, stream.state());
42+
LOG.info("Starter konsumering av topics={}", kcm.topicNames());
43+
kcm.start((t, e) -> LOG.error("{} :: Caught exception in stream, exiting", t, e));
8144
}
8245

8346
@Override
8447
public void stop() {
85-
LOG.info("Starter shutdown av topic={}, tilstand={} med 10 sekunder timeout", topic, stream.state());
86-
stream.close(Duration.of(30, ChronoUnit.SECONDS));
87-
LOG.info("Shutdown av topic={}, tilstand={} med 10 sekunder timeout", topic, stream.state());
48+
LOG.info("Starter shutdown av topics={} med 10 sekunder timeout", kcm.topicNames());
49+
kcm.stop();
8850
}
8951
}

‎domenetjenester/vedtak/src/main/java/no/nav/foreldrepenger/abakus/vedtak/kafka/VedtaksHendelseHåndterer.java‎

Lines changed: 23 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,10 @@
55
import jakarta.inject.Inject;
66
import jakarta.transaction.Transactional;
77

8+
import no.nav.foreldrepenger.konfig.KonfigVerdi;
9+
10+
import no.nav.vedtak.felles.integrasjon.kafka.KafkaMessageHandler;
11+
812
import org.slf4j.Logger;
913
import org.slf4j.LoggerFactory;
1014

@@ -15,24 +19,38 @@
1519
@ApplicationScoped
1620
@ActivateRequestContext
1721
@Transactional
18-
public class VedtaksHendelseHåndterer {
22+
public class VedtaksHendelseHåndterer implements KafkaMessageHandler.KafkaStringMessageHandler{
1923

2024
private static final Logger LOG = LoggerFactory.getLogger(VedtaksHendelseHåndterer.class);
2125
private ProsessTaskTjeneste taskTjeneste;
26+
private String topicName;
2227

2328
public VedtaksHendelseHåndterer() {
2429
// CDI
2530
}
2631

2732
@Inject
28-
public VedtaksHendelseHåndterer(ProsessTaskTjeneste taskTjeneste) {
33+
public VedtaksHendelseHåndterer(@KonfigVerdi(value = "kafka.fattevedtak.topic", defaultVerdi = "teamforeldrepenger.familie-vedtakfattet-v1") String topicName,
34+
ProsessTaskTjeneste taskTjeneste) {
35+
this.topicName = topicName;
2936
this.taskTjeneste = taskTjeneste;
3037
}
3138

32-
void handleMessage(String key, String payload) {
33-
LOG.debug("Mottatt ytelse-vedtatt hendelse med key='{}', payload={}", key, payload);
34-
var data = ProsessTaskDataBuilder.forProsessTask(LagreVedtakTask.class).medProperty(LagreVedtakTask.KEY, key).medPayload(payload);
39+
@Override
40+
public void handleRecord(String key, String value) {
41+
LOG.debug("Mottatt ytelse-vedtatt hendelse med key='{}', payload={}", key, value);
42+
var data = ProsessTaskDataBuilder.forProsessTask(LagreVedtakTask.class).medProperty(LagreVedtakTask.KEY, key).medPayload(value);
3543

3644
taskTjeneste.lagre(data.build());
3745
}
46+
47+
@Override
48+
public String topic() {
49+
return topicName;
50+
}
51+
52+
@Override
53+
public String groupId() {
54+
return "fpabakus"; // Hold konstant pga offset commit
55+
}
3856
}

‎felles/pom.xml‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -39,10 +39,6 @@
3939
<groupId>no.nav.foreldrepenger.felles</groupId>
4040
<artifactId>felles-util</artifactId>
4141
</dependency>
42-
<dependency>
43-
<groupId>org.apache.kafka</groupId>
44-
<artifactId>kafka-streams</artifactId>
45-
</dependency>
4642
<dependency>
4743
<groupId>no.nav.foreldrepenger.felles</groupId>
4844
<artifactId>felles-abac-kontekst</artifactId>

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@
3535

3636
<kontrakt.java.version>21</kontrakt.java.version>
3737

38-
<felles.version>7.0.7</felles.version>
38+
<felles.version>7.0.8</felles.version>
3939
<prosesstask.version>5.0.5</prosesstask.version>
4040
<kontrakter.version>9.1.8</kontrakter.version>
4141
<tidsserie.version>2.7.1</tidsserie.version>

0 commit comments

Comments
 (0)