Skip to content

Switching to 4.3.1 for the Kafka client libs #3985

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
shroffk merged 1 commit into from
Oct 26, 2026
+93 −156
Merged

Switching to 4.3.1 for the Kafka client libs #3985

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 5 additions & 12 deletions app/alarm/model/src/main/java/org/phoebus/applications/alarm/talk/TalkClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,13 +83,6 @@ public void addListener(final TalkClientListener listener)
listeners.add(listener);
}

/** @param listener - Listener to remove */
public void removeListener(final TalkClientListener listener)
{
if (! listeners.remove(listener))
throw new IllegalStateException("Unknown listener");
}

/** Start client
* @see #shutdown()
*/
Expand Down Expand Up @@ -127,7 +120,7 @@ private void run()
/** Perform one check for updates */
private void checkUpdates()
{
final ConsumerRecords<String, String> records = consumer.poll(100);
final ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (final ConsumerRecord<String, String> record : records)
{
String path = record.key();
Expand All @@ -145,9 +138,9 @@ private void checkUpdates()

// Extract the message info from the JSON and notify the listeners.

final String severity = jn.get(JsonTags.SEVERITY).textValue();
final String severity = jn.get(JsonTags.SEVERITY).stringValue();
final boolean standout = jn.get(JsonTags.STANDOUT).asBoolean();
final String message = jn.get(JsonTags.TALK).textValue();
final String message = jn.get(JsonTags.TALK).stringValue();

try
{
Expand All @@ -166,11 +159,11 @@ private void checkUpdates()
private Optional<Instant> nextDisconnectedAnnunciation = Optional.empty(); // When alarm server is disconnected: point in time for next annunciation of disconnection.
private final Duration disconnectionAnnunciationPeriod = Duration.ofMillis(AlarmSystem.nag_period_ms);
private final Duration idleTimeoutDuration = Duration.ofMillis(AlarmSystem.idle_timeout_ms).multipliedBy(3);
private AtomicReference<Instant> lastReceivedUpdateFromAlarmServer = new AtomicReference<>(Instant.now());
private final AtomicReference<Instant> lastReceivedUpdateFromAlarmServer = new AtomicReference<>(Instant.now());

private void updateHeartbeatTimestampLoop() {
while (running.get()) {
final ConsumerRecords<String, String> records = heartbeatConsumer.poll(100);
final ConsumerRecords<String, String> records = heartbeatConsumer.poll(Duration.ofMillis(100));
if (!records.isEmpty()) {
lastReceivedUpdateFromAlarmServer.set(Instant.now());
}
Expand Down
2 changes: 1 addition & 1 deletion pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@
<apache.commons.math.version>3.6.1</apache.commons.math.version>
<junit.version>5.8.2</junit.version>
<elasticsearch.version>9.4.4</elasticsearch.version>
<kafka.version>3.9.1</kafka.version>
<kafka.version>4.3.1</kafka.version>
<!--<maven.repo.local>${project.build.directory}/.m2</maven.repo.local> -->
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
Expand Down
28 changes: 11 additions & 17 deletions services/alarm-config-logger/src/main/java/org/phoebus/alarm/logging/AlarmConfigLogger.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,9 +32,9 @@
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.Record;
import org.eclipse.jgit.api.Git;
import org.eclipse.jgit.api.PushCommand;
import org.eclipse.jgit.api.RemoteRemoveCommand;
Expand Down Expand Up @@ -65,7 +65,7 @@

private final String topic;
private final String remoteLocation;
private Properties props;
private final Properties props;

private final File root;
private final String group_id;
Expand Down Expand Up @@ -155,7 +155,7 @@

writeAlarmModel();
try (Consumer<String, String> consumer = new KafkaConsumer<>(props,
Serdes.String().deserializer(), Serdes.String().deserializer());) {
Serdes.String().deserializer(), Serdes.String().deserializer())) {

// Rewind whenever assigned to partition
final ConsumerRebalanceListener crl = new ConsumerRebalanceListener() {
Expand Down Expand Up @@ -193,13 +193,7 @@
StreamsBuilder builder = new StreamsBuilder();

KStream<String, String> alarms = builder.stream(topic, Consumed.with(Serdes.String(), Serdes.String()));
alarms.process(new ProcessorSupplier<>() {
@Override
public Processor<String, String> get() {
return new ProcessAlarmConfigMessage();
}
});

alarms.process(() -> new ProcessAlarmConfigMessage());
Topology topology = builder.build();
logger.config(topology.describe().toString());
streams = new KafkaStreams(topology, props);
Expand Down Expand Up @@ -306,15 +300,15 @@
}
}

private class ProcessAlarmConfigMessage implements Processor<String, String> {
private class ProcessAlarmConfigMessage implements Processor<String, String, String, String> {

@Override
public void init(ProcessorContext context) {
public void init(ProcessorContext<String, String> context) {

Check failure on line 306 in services/alarm-config-logger/src/main/java/org/phoebus/alarm/logging/AlarmConfigLogger.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Add a nested comment explaining why this method is empty, throw an UnsupportedOperationException or complete the implementation. 


See more on https://sonarcloud.io/project/issues?id=ControlSystemStudio_phoebus&issues=AaFwU82cYghEOE8bdHho&open=AaFwU82cYghEOE8bdHho&pullRequest=3985
}

@Override
public synchronized void process(String key, String value) {
processAlarmConfigMessages(key, value, true);
public synchronized void process(Record<String, String> record) {

Check warning on line 310 in services/alarm-config-logger/src/main/java/org/phoebus/alarm/logging/AlarmConfigLogger.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Rename this variable to not match a restricted identifier. 


See more on https://sonarcloud.io/project/issues?id=ControlSystemStudio_phoebus&issues=AaFwU82cYghEOE8bdHhp&open=AaFwU82cYghEOE8bdHhp&pullRequest=3985
processAlarmConfigMessages(record.key(), record.value(), true);
}

@Override
Expand All @@ -331,7 +325,7 @@
}
File node_info = new File(node, "config.xml");
try (OutputStream fo = Files.newOutputStream(node_info.toPath());
XmlModelWriter modelWriter = new XmlModelWriter(fo);) {
XmlModelWriter modelWriter = new XmlModelWriter(fo)) {
modelWriter.write(model.getRoot());
} catch (Exception e) {
logger.log(Level.WARNING, "Alarm config logging failed to dump the alarm configuration to config.xml", e);
Expand Down
55 changes: 19 additions & 36 deletions services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmCmdLogger.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,52 +13,47 @@
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Transformer;
import org.apache.kafka.streams.kstream.TransformerSupplier;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.TimestampExtractor;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.ProcessorSupplier;
import org.apache.kafka.streams.processor.api.Record;
import org.phoebus.applications.alarm.client.KafkaHelper;
import org.phoebus.applications.alarm.messages.AlarmCommandMessage;
import org.phoebus.applications.alarm.messages.MessageParser;
import org.phoebus.framework.preferences.PreferencesReader;
import org.phoebus.util.indexname.IndexNameHelper;

/**
* A Runnable which consumes the alarm command messages and records them to an
* elastic index.
* elastic index.
*
* @author Kunal Shroff
*
*/
public class AlarmCmdLogger implements Runnable {

private static final PreferencesReader prefs = new PreferencesReader(AlarmLoggingService.class, "/application.properties");

private static final String INDEX_FORMAT = "_alarms_cmd";
private final String topic;
private final Serde<AlarmCommandMessage> alarmCommandMessageSerde;


private IndexNameHelper indexNameHelper;

private volatile boolean shouldReconnect = true;
private volatile KafkaStreams currentStreams = null;
private final long reconnectDelayMs;
private Thread shutdownHook = null;

/**
* Create a alarm command message logger for the given topic.
* Create a alarm command message logger for the given topic.
* This runnable will create the kafka streams for the given alarm messages which match the format 'topicCommand'
* @param topic the alarm topic
* @throws Exception - parsing the alarm command messages
*/
public AlarmCmdLogger(String topic) throws Exception {
public AlarmCmdLogger(String topic) {
super();
this.topic = topic;

Expand Down Expand Up @@ -88,7 +83,7 @@
}

// Register shutdown hook once before retry loop
shutdownHook = new Thread("streams-" + topic + "-alarm-cmd-shutdown-hook") {
Thread shutdownHook = new Thread("streams-" + topic + "-alarm-cmd-shutdown-hook") {
@Override
public void run() {
logger.info("Shutdown hook triggered for topic " + topic);
Expand Down Expand Up @@ -140,7 +135,7 @@
logger.info("Alarm cmd logger for topic " + topic + " has shut down");
}

private void startKafkaStreams(Properties props) throws Exception {

Check warning on line 138 in services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmCmdLogger.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Replace generic exceptions with specific library exceptions or a custom exception. 


See more on https://sonarcloud.io/project/issues?id=ControlSystemStudio_phoebus&issues=AaFwU8scYghEOE8bdHhf&open=AaFwU8scYghEOE8bdHhf&pullRequest=3985
logger.info("Attempting to start Kafka Streams for topic " + topic);

Properties kafkaProps = KafkaHelper.loadPropsFromFile(props.getProperty("kafka_properties",""));
Expand All @@ -160,41 +155,29 @@
}
}));

KStream<String, AlarmCommandMessage> timeStampedAlarms = alarms.transform(new TransformerSupplier<>() {

alarms.process(new ProcessorSupplier<String, AlarmCommandMessage, String, AlarmCommandMessage>() {
@Override
public Transformer<String, AlarmCommandMessage, KeyValue<String, AlarmCommandMessage>> get() {
return new Transformer<>() {
private ProcessorContext context;
public Processor<String, AlarmCommandMessage, String, AlarmCommandMessage> get() {
return new Processor<>() {
private ProcessorContext<String, AlarmCommandMessage> context;

@Override
public void init(ProcessorContext context) {
this.context = context;
}

@Override
public KeyValue<String, AlarmCommandMessage> transform(String key, AlarmCommandMessage value) {
key = key.replace("\\", "");
value.setConfig(key);
value.setMessage_time(Instant.ofEpochMilli(context.timestamp()));
return new KeyValue<>(key, value);
}

@Override
public void close() {

public void process(Record<String, AlarmCommandMessage> record) {

Check warning on line 170 in services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmCmdLogger.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Rename this variable to not match a restricted identifier. 


See more on https://sonarcloud.io/project/issues?id=ControlSystemStudio_phoebus&issues=AaFwU8scYghEOE8bdHhg&open=AaFwU8scYghEOE8bdHhg&pullRequest=3985
String key = record.key().replace("\\", "");
record.value().setConfig(key);
record.value().setMessage_time(Instant.ofEpochMilli(context.currentStreamTimeMs()));
String topic_name = indexNameHelper.getIndexName(record.value().getMessage_time());

Check warning on line 174 in services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmCmdLogger.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Rename this local variable to match the regular expression '^[a-z][a-zA-Z0-9]*$'. 


See more on https://sonarcloud.io/project/issues?id=ControlSystemStudio_phoebus&issues=AaFwU8scYghEOE8bdHhe&open=AaFwU8scYghEOE8bdHhe&pullRequest=3985
ElasticClientHelper.getInstance().indexAlarmCmdDocument(topic_name, record.value());
}

};
}
});

// Commit to elastic
timeStampedAlarms.foreach((k, v) -> {
String topic_name = indexNameHelper.getIndexName(v.getMessage_time());
ElasticClientHelper.getInstance().indexAlarmCmdDocument(topic_name, v);
});

final KafkaStreams streams = new KafkaStreams(builder.build(), kafkaProps);

// Store reference for cleanup (volatile ensures visibility across threads)
Expand Down
Loading
Loading