diff --git a/app/alarm/model/src/main/java/org/phoebus/applications/alarm/talk/TalkClient.java b/app/alarm/model/src/main/java/org/phoebus/applications/alarm/talk/TalkClient.java index 46e17107d9..fce82cb1a3 100644 --- a/app/alarm/model/src/main/java/org/phoebus/applications/alarm/talk/TalkClient.java +++ b/app/alarm/model/src/main/java/org/phoebus/applications/alarm/talk/TalkClient.java @@ -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() */ @@ -127,7 +120,7 @@ private void run() /** Perform one check for updates */ private void checkUpdates() { - final ConsumerRecords records = consumer.poll(100); + final ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); for (final ConsumerRecord record : records) { String path = record.key(); @@ -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 { @@ -166,11 +159,11 @@ private void checkUpdates() private Optional 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 lastReceivedUpdateFromAlarmServer = new AtomicReference<>(Instant.now()); + private final AtomicReference lastReceivedUpdateFromAlarmServer = new AtomicReference<>(Instant.now()); private void updateHeartbeatTimestampLoop() { while (running.get()) { - final ConsumerRecords records = heartbeatConsumer.poll(100); + final ConsumerRecords records = heartbeatConsumer.poll(Duration.ofMillis(100)); if (!records.isEmpty()) { lastReceivedUpdateFromAlarmServer.set(Instant.now()); } diff --git a/pom.xml b/pom.xml index 060b046f3d..45c50fd4f0 100644 --- a/pom.xml +++ b/pom.xml @@ -83,7 +83,7 @@ 3.6.1 5.8.2 9.4.4 - 3.9.1 + 4.3.1 UTF-8 UTF-8 diff --git a/services/alarm-config-logger/src/main/java/org/phoebus/alarm/logging/AlarmConfigLogger.java b/services/alarm-config-logger/src/main/java/org/phoebus/alarm/logging/AlarmConfigLogger.java index 0276df04bd..f5230b6d5a 100644 --- a/services/alarm-config-logger/src/main/java/org/phoebus/alarm/logging/AlarmConfigLogger.java +++ b/services/alarm-config-logger/src/main/java/org/phoebus/alarm/logging/AlarmConfigLogger.java @@ -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; @@ -65,7 +65,7 @@ public class AlarmConfigLogger implements Runnable { private final String topic; private final String remoteLocation; - private Properties props; + private final Properties props; private final File root; private final String group_id; @@ -155,7 +155,7 @@ private void initialize() { writeAlarmModel(); try (Consumer 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() { @@ -193,13 +193,7 @@ public void run() { StreamsBuilder builder = new StreamsBuilder(); KStream alarms = builder.stream(topic, Consumed.with(Serdes.String(), Serdes.String())); - alarms.process(new ProcessorSupplier<>() { - @Override - public Processor get() { - return new ProcessAlarmConfigMessage(); - } - }); - + alarms.process(() -> new ProcessAlarmConfigMessage()); Topology topology = builder.build(); logger.config(topology.describe().toString()); streams = new KafkaStreams(topology, props); @@ -306,15 +300,15 @@ private synchronized void syncAlarmConfigRepository(ConsumerRecords { + private class ProcessAlarmConfigMessage implements Processor { @Override - public void init(ProcessorContext context) { + public void init(ProcessorContext context) { } @Override - public synchronized void process(String key, String value) { - processAlarmConfigMessages(key, value, true); + public synchronized void process(Record record) { + processAlarmConfigMessages(record.key(), record.value(), true); } @Override @@ -331,7 +325,7 @@ private synchronized void writeAlarmModel() { } 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); diff --git a/services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmCmdLogger.java b/services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmCmdLogger.java index 4d846a33db..1728f5088a 100644 --- a/services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmCmdLogger.java +++ b/services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmCmdLogger.java @@ -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 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; @@ -88,7 +83,7 @@ public void run() { } // 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); @@ -160,12 +155,11 @@ public long extract(ConsumerRecord record, long previousTimestam } })); - KStream timeStampedAlarms = alarms.transform(new TransformerSupplier<>() { - + alarms.process(new ProcessorSupplier() { @Override - public Transformer> get() { - return new Transformer<>() { - private ProcessorContext context; + public Processor get() { + return new Processor<>() { + private ProcessorContext context; @Override public void init(ProcessorContext context) { @@ -173,28 +167,17 @@ public void init(ProcessorContext context) { } @Override - public KeyValue 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 record) { + 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()); + 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) diff --git a/services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmMessageLogger.java b/services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmMessageLogger.java index d733cb27d3..e2ac1e9977 100644 --- a/services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmMessageLogger.java +++ b/services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmMessageLogger.java @@ -23,10 +23,11 @@ import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Named; -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.AlarmSystemConstants; import org.phoebus.applications.alarm.client.KafkaHelper; import org.phoebus.applications.alarm.messages.AlarmConfigMessage; @@ -51,7 +52,6 @@ public class AlarmMessageLogger implements Runnable { private volatile boolean shouldReconnect = true; private volatile KafkaStreams currentStreams = null; private final long reconnectDelayMs; - private Thread shutdownHook = null; /** * Create a alarm logger for the alarm messages (both state and configuration) @@ -91,7 +91,7 @@ public void run() { } // Register shutdown hook once before retry loop - shutdownHook = new Thread("streams-"+topic+"-alarm-messages-shutdown-hook") { + Thread shutdownHook = new Thread("streams-"+topic+"-alarm-messages-shutdown-hook") { @Override public void run() { logger.info("Shutdown hook triggered for topic " + topic); @@ -169,20 +169,15 @@ private void startKafkaStreams(Properties props) throws Exception { KStream alarms = builder.stream(topic, Consumed.with(Serdes.String(), alarmMessageSerde).withTimestampExtractor(new TimestampExtractor() { - @Override public long extract(ConsumerRecord record, long previousTimestamp) { return record.timestamp(); } })); - alarms = alarms.filter((k, v) -> { - return v != null; - }); + alarms = alarms.filter((k, v) -> v != null); alarms = alarms.map((key, value) -> { -// logger.config("Processing alarm message with key : " + key != null ? key -// : "null" + " " + value != null ? value.toString() : "null"); value.setKey(key); return new KeyValue<>(key, value); }); @@ -251,62 +246,43 @@ public long extract(ConsumerRecord record, long previousTimestam } private void processAlarmStateStream(KStream alarmStateBranch) { - - KStream transformedAlarms = alarmStateBranch - .transform(new TransformerSupplier<>() { - + alarmStateBranch.process(new ProcessorSupplier() { + @Override + public Processor get() { + return new Processor<>() { + private ProcessorContext context; @Override - public Transformer> get() { - return new Transformer<>() { - private ProcessorContext context; - - @Override - public KeyValue transform(String key, AlarmMessage value) { - key = key.replace("\\", ""); - AlarmStateMessage newValue = value.getAlarmStateMessage(); - Matcher matcher = pattern.matcher(key); - newValue.setConfig(key); - matcher.find(); - String[] tokens = AlarmTreePath.splitPath(key); - final String pv = tokens[tokens.length - 1]; - newValue.setPv(pv); - - newValue.setMessage_time(Instant.ofEpochMilli(context.timestamp())); - return new KeyValue<>(key, newValue); - } - - @Override - public void init(ProcessorContext context) { - this.context = context; - } - - @Override - public void close() { - // TODO Auto-generated method stub - - } - }; + public void process(Record record) { + AlarmStateMessage newValue = record.value().getAlarmStateMessage(); + if(!newValue.isLeaf()){ + return; + } + String key = record.key().replace("\\", ""); + Matcher matcher = pattern.matcher(key); + newValue.setConfig(key); + matcher.find(); + String[] tokens = AlarmTreePath.splitPath(key); + final String pv = tokens[tokens.length - 1]; + newValue.setPv(pv); + newValue.setMessage_time(Instant.ofEpochMilli(context.currentStreamTimeMs())); + ElasticClientHelper.getInstance().indexAlarmStateDocuments(stateIndexNameHelper.getIndexName(newValue.getMessage_time()), newValue); } - }); - KStream filteredAlarms = transformedAlarms.filter((k, v) -> { - return v != null ? v.isLeaf() : false; - }); - - // Commit to elastic - filteredAlarms.foreach((k, v) -> { - ElasticClientHelper.getInstance().indexAlarmStateDocuments(stateIndexNameHelper.getIndexName(v.getMessage_time()), v); + @Override + public void init(ProcessorContext context) { + this.context = context; + } + }; + } }); - } private void processAlarmConfigurationStream(KStream alarmConfigBranch) { - KStream alarmConfigMessages = alarmConfigBranch.transform(new TransformerSupplier<>() { - + alarmConfigBranch.process(new ProcessorSupplier() { @Override - public Transformer> get() { - return new Transformer<>() { - private ProcessorContext context; + public Processor get() { + return new Processor<>() { + private ProcessorContext context; @Override public void init(ProcessorContext context) { @@ -314,32 +290,18 @@ public void init(ProcessorContext context) { } @Override - public KeyValue transform(String key, AlarmMessage value) { + public void process(Record record) { - key = key.replace("\\", ""); - if (value != null) { - AlarmConfigMessage newValue = value.getAlarmConfigMessage(); + String key = record.key().replace("\\", ""); + if (record.value() != null) { + AlarmConfigMessage newValue = record.value().getAlarmConfigMessage(); newValue.setConfig(key); - newValue.setMessage_time(Instant.ofEpochMilli(context.timestamp())); - return new KeyValue<>(key, newValue); - } else { - return null; + newValue.setMessage_time(Instant.ofEpochMilli(context.currentStreamTimeMs())); + ElasticClientHelper.getInstance().indexAlarmConfigDocuments(configIndexNameHelper.getIndexName(newValue.getMessage_time()), newValue); } } - - @Override - public void close() { - - } - }; } }); - - // Commit to elastic - alarmConfigMessages.foreach((k, v) -> { - ElasticClientHelper.getInstance().indexAlarmConfigDocuments(configIndexNameHelper.getIndexName(v.getMessage_time()), v); - }); } - } diff --git a/services/alarm-server/src/main/java/org/phoebus/applications/alarm/aggregate/AggregateTopics.java b/services/alarm-server/src/main/java/org/phoebus/applications/alarm/aggregate/AggregateTopics.java index 5bfa91765d..5e1b8d3699 100644 --- a/services/alarm-server/src/main/java/org/phoebus/applications/alarm/aggregate/AggregateTopics.java +++ b/services/alarm-server/src/main/java/org/phoebus/applications/alarm/aggregate/AggregateTopics.java @@ -17,6 +17,7 @@ import java.util.logging.Logger; import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler; import org.phoebus.applications.alarm.AlarmSystem; import org.phoebus.applications.alarm.client.KafkaHelper; import org.phoebus.applications.alarm.server.CreateTopics; @@ -27,7 +28,7 @@ @SuppressWarnings("nls") public class AggregateTopics { - private Logger logger = Logger.getLogger(this.getClass().getPackageName()); + private final Logger logger = Logger.getLogger(this.getClass().getPackageName()); private String kafka_servers = "localhost:9092"; private String config = "Accelerator"; private String kafka_properties = ""; @@ -43,12 +44,12 @@ private AggregateTopics(String[] args) longTerm = config + AlarmSystem.LONG_TERM_TOPIC_SUFFIX; if (createTopic) { - logger.info("Discovering and creating topics in " + topics.toString()); + logger.info("Discovering and creating topics in " + topics); CreateTopics.discoverAndCreateTopics(kafka_servers, false, List.of(longTerm), kafka_properties); } logger.info("server:\"" + kafka_servers + "\", config: \"" + config + "\""); - logger.info("topics: " + topics.toString()); + logger.info("topics: " + topics); aggregateStream = createStream(); logger.info("Starting stream aggregation."); @@ -56,7 +57,7 @@ private AggregateTopics(String[] args) try ( - BufferedReader bufReader = new BufferedReader(new InputStreamReader(System.in)); + BufferedReader bufReader = new BufferedReader(new InputStreamReader(System.in)) ) { System.out.println("Type \"exit\" to stop."); @@ -159,10 +160,7 @@ else if (arg.equals("-kafka_properties")) private List createTopicList() { - List list = List.of(config, - config + AlarmSystem.COMMAND_TOPIC_SUFFIX, - config + AlarmSystem.TALK_TOPIC_SUFFIX); - return list; + return List.of(config, config + AlarmSystem.COMMAND_TOPIC_SUFFIX, config + AlarmSystem.TALK_TOPIC_SUFFIX); } private KafkaStreams createStream() @@ -173,16 +171,17 @@ private KafkaStreams createStream() kafka_properties); // Log any uncaught exceptions. - stream.setUncaughtExceptionHandler((Thread thread, Throwable throwable) -> { + stream.setUncaughtExceptionHandler(throwable -> { logger.log(Level.WARNING, "Kafka Streams Error.", throwable); - }); + return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_APPLICATION; + }); // Catch control-c and shutdown stream beforehand. Runtime.getRuntime().addShutdownHook(new Thread(stream::close)); return stream; } - public static void main(String[] args) + static void main(String[] args) { new AggregateTopics(args); } diff --git a/services/save-and-restore/pom.xml b/services/save-and-restore/pom.xml index 5a81b8a8a1..887557d26b 100644 --- a/services/save-and-restore/pom.xml +++ b/services/save-and-restore/pom.xml @@ -80,6 +80,12 @@ 7.0.0-SNAPSHOT + + org.phoebus + core-pv-jackie + 7.0.0-SNAPSHOT + + org.phoebus core-websocket-common