<persistence-unit name="integrityAuditPU" transaction-type="RESOURCE_LOCAL">
<!-- For operational use -->
+ <provider>org.hibernate.jpa.HibernatePersistenceProvider</provider>
<class>org.onap.policy.common.ia.jpa.IntegrityAuditEntity</class>
<shared-cache-mode>NONE</shared-cache-mode>
<properties>
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/persistence http://xmlns.jcp.org/xml/ns/persistence/persistence_2_1.xsd">
<persistence-unit name="schemaPU" transaction-type="RESOURCE_LOCAL">
<!-- Limited use for generating the DB and schema files for imtest DB - uses hibernate -->
+ <provider>org.hibernate.jpa.HibernatePersistenceProvider</provider>
<class>org.onap.policy.common.im.jpa.ImTestEntity</class>
<class>org.onap.policy.common.im.jpa.StateManagementEntity</class>
<class>org.onap.policy.common.im.jpa.ForwardProgressEntity</class>
* policy-endpoints
* ================================================================================
* Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
*
* @return true if the send operation succeeded, false otherwise
* @throws IllegalArgumentException an invalid message has been provided
- * @throws IllegalStateException the entity is in an state that prevents
+ * @throws IllegalStateException the entity is in a state that prevents
* it from sending messages, for example, locked or stopped.
*/
- public boolean send(String message);
+ boolean send(String message);
}
* policy-endpoints
* ================================================================================
* Copyright (C) 2017, 2019 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
*
* @param partitionKey the partition key
*/
- public void setPartitionKey(String partitionKey);
+ void setPartitionKey(String partitionKey);
/**
* Return the partition key in used by the system to publish messages.
*
* @return the partition key
*/
- public String getPartitionKey();
+ String getPartitionKey();
}
/*
* ============LICENSE_START=======================================================
- * Copyright (C) 2022 Nordix Foundation.
+ * Copyright (C) 2022-2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
/**
* Logger.
*/
- private static Logger logger = LoggerFactory.getLogger(IndexedKafkaTopicSinkFactory.class);
+ private static final Logger logger = LoggerFactory.getLogger(IndexedKafkaTopicSinkFactory.class);
/**
* KAFKA Topic Name Index.
List<KafkaTopicSink> newKafkaTopicSinks = new ArrayList<>();
synchronized (this) {
for (String topic : COMMA_SPACE_PAT.split(writeTopics)) {
- addTopic(newKafkaTopicSinks, topic, properties);
+ addTopic(newKafkaTopicSinks, topic.toLowerCase(), properties);
}
return newKafkaTopicSinks;
}
String topicPrefix = PolicyEndPointProperties.PROPERTY_KAFKA_SINK_TOPICS + "." + topic;
var props = new PropertyUtils(properties, topicPrefix,
- (name, value, ex) -> logger.warn("{}: {} {} is in invalid format for topic {} ", this, name, value, topic));
+ (name, value, ex) -> logger.warn("{}: {} {} is in invalid format for topic sink {} ",
+ this, name, value, topic));
String servers = properties.getProperty(topicPrefix + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
if (StringUtils.isBlank(servers)) {
/**
* Logger.
*/
- private static Logger logger = LoggerFactory.getLogger(IndexedKafkaTopicSourceFactory.class);
+ private static final Logger logger = LoggerFactory.getLogger(IndexedKafkaTopicSourceFactory.class);
/**
* KAFKA Topic Name Index.
List<KafkaTopicSource> newKafkaTopicSources = new ArrayList<>();
synchronized (this) {
for (String topic : COMMA_SPACE_PAT.split(readTopics)) {
- addTopic(newKafkaTopicSources, topic, properties);
+ addTopic(newKafkaTopicSources, topic.toLowerCase(), properties);
}
}
return newKafkaTopicSources;
String topicPrefix = PolicyEndPointProperties.PROPERTY_KAFKA_SOURCE_TOPICS + "." + topic;
var props = new PropertyUtils(properties, topicPrefix,
- (name, value, ex) -> logger.warn("{}: {} {} is in invalid format for topic {} ", this, name, value, topic));
+ (name, value, ex) -> logger.warn("{}: {} {} is in invalid format for topic source {} ",
+ this, name, value, topic));
String servers = properties.getProperty(topicPrefix + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
if (StringUtils.isBlank(servers)) {
- logger.error("{}: no KAFKA servers configured for sink {}", this, topic);
+ logger.error("{}: no KAFKA servers configured for source {}", this, topic);
return;
}
/*
* ============LICENSE_START=======================================================
- * Copyright (C) 2022 Nordix Foundation.
+ * Copyright (C) 2022-2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
public interface KafkaTopicSourceFactory {
/**
- * Creates an Kafka Topic Source based on properties files.
+ * Creates a Kafka Topic Source based on properties files.
*
* @param properties Properties containing initialization values
*
- * @return an Kafka Topic Source
+ * @return a Kafka Topic Source
* @throws IllegalArgumentException if invalid parameters are present
*/
List<KafkaTopicSource> build(Properties properties);
* Instantiates a new Kafka Topic Source.
*
* @param busTopicParams parameters object
- * @return an Kafka Topic Source
+ * @return a Kafka Topic Source
*/
KafkaTopicSource build(BusTopicParams busTopicParams);
* @param servers list of servers
* @param topic topic name
*
- * @return an Kafka Topic Source
+ * @return a Kafka Topic Source
* @throws IllegalArgumentException if invalid parameters are present
*/
KafkaTopicSource build(List<String> servers, String topic);
/**
- * Destroys an Kafka Topic Source based on a topic.
+ * Destroys a Kafka Topic Source based on a topic.
*
* @param topic topic name
* @throws IllegalArgumentException if invalid parameters are present
void destroy();
/**
- * Gets an Kafka Topic Source based on topic name.
+ * Gets a Kafka Topic Source based on topic name.
*
* @param topic the topic name
- * @return an Kafka Topic Source with topic name
+ * @return a Kafka Topic Source with topic name
* @throws IllegalArgumentException if an invalid topic is provided
* @throws IllegalStateException if the Kafka Topic Source is an incorrect state
*/
* ================================================================================
* Copyright (C) 2018 Samsung Electronics Co., Ltd. All rights reserved.
* Modifications Copyright (C) 2018-2019, 2021 AT&T Intellectual Property. All rights reserved.
- * Modifications Copyright (C) 2019 Nordix Foundation.
+ * Modifications Copyright (C) 2019, 2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
return additionalProps != null;
}
+ public void setEffectiveTopic(String effectiveTopic) {
+ this.effectiveTopic = topicToLowerCase(effectiveTopic);
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topicToLowerCase(topic);
+ }
+
+ public String getEffectiveTopic() {
+ return topicToLowerCase(effectiveTopic);
+ }
+
+ public String getTopic() {
+ return topicToLowerCase(topic);
+ }
+
+ private String topicToLowerCase(String topic) {
+ return (topic == null || topic.isEmpty()) ? topic : topic.toLowerCase();
+ }
+
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public static class TopicParamsBuilder {
}
public TopicParamsBuilder topic(String topic) {
- this.params.topic = topic;
+ this.params.setTopic(topic);
return this;
}
public TopicParamsBuilder effectiveTopic(String effectiveTopic) {
- this.params.effectiveTopic = effectiveTopic;
+ this.params.setEffectiveTopic(effectiveTopic);
return this;
}
this.params.serializationProvider = serializationProvider;
return this;
}
-
}
}
* Copyright (C) 2017-2021 AT&T Intellectual Property. All rights reserved.
* Modifications Copyright (C) 2018-2019 Samsung Electronics Co., Ltd.
* Modifications Copyright (C) 2020 Bell Canada. All rights reserved.
+ * Modifications Copyright (C) 2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
this.stop();
}
- @Override
- protected boolean anyNullOrEmpty(String... args) {
- for (String arg : args) {
- if (arg == null || arg.isEmpty()) {
- return true;
- }
- }
-
- return false;
- }
-
- @Override
- protected boolean allNullOrEmpty(String... args) {
- for (String arg : args) {
- if (!(arg == null || arg.isEmpty())) {
- return false;
- }
- }
-
- return true;
- }
-
@Override
public String toString() {
return "InlineBusTopicSink [partitionId=" + partitionKey + ", alive=" + alive + ", publisher=" + publisher
/*
* ============LICENSE_START=======================================================
- * Copyright (C) 2022 Nordix Foundation.
+ * Copyright (C) 2022-2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
/**
* Logger.
*/
- private static Logger logger = LoggerFactory.getLogger(InlineKafkaTopicSink.class);
+ private static final Logger logger = LoggerFactory.getLogger(InlineKafkaTopicSink.class);
- protected Map<String, String> additionalProps = null;
+ protected Map<String, String> additionalProps;
/**
- * Argument-based KAFKA Topic Writer instantiation. BusTopicParams contains below mentioned
+ * Argument-based KAFKA Topic Writer instantiation. BusTopicParams contains the below
* attributes.
*
* <p>servers list of KAFKA servers available for publishing
/*
* ============LICENSE_START=======================================================
- * Copyright (C) 2022 Nordix Foundation.
+ * Copyright (C) 2022-2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import org.onap.policy.common.endpoints.event.comm.bus.KafkaTopicSource;
/**
- * This topic source implementation specializes in reading messages over an Kafka Bus topic source and
+ * This topic source implementation specializes in reading messages over a Kafka Bus topic source and
* notifying its listeners.
*/
public class SingleThreadedKafkaTopicSource extends SingleThreadedBusTopicSource implements KafkaTopicSource {
}
this.servers = servers;
- this.topic = topic;
- this.effectiveTopic = effectiveTopicCopy;
+ this.topic = topic.toLowerCase();
+ this.effectiveTopic = effectiveTopicCopy.toLowerCase();
}
@Override
* ONAP
* ================================================================================
* Copyright (C) 2020-2021 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
package org.onap.policy.common.endpoints.event.comm.client;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import lombok.Getter;
+import org.jetbrains.annotations.NotNull;
import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
private final CommInfrastructure sourceTopicCommInfrastructure;
/**
- * Used when checking whether or not a message sent on the sink topic can be received
+ * Used when checking whether a message sent on the sink topic can be received
* on the source topic. When a matching message is received on the incoming topic,
- * {@code true} is placed on the queue. If {@link #stop()} is called or the waiting
+ * {@code true} is placed on the queue. If {@link #stopWaiting()} is called or the waiting
* thread is interrupted, then {@code false} is placed on the queue. Whenever a value
* is pulled from the queue, it is immediately placed back on the queue.
*/
* @throws BidirectionalTopicClientException if either topic does not exist
*/
public BidirectionalTopicClient(String sinkTopic, String sourceTopic) throws BidirectionalTopicClientException {
- this.sinkTopic = sinkTopic;
- this.sourceTopic = sourceTopic;
+ this.sinkTopic = sinkTopic.toLowerCase();
+ this.sourceTopic = sourceTopic.toLowerCase();
// init sinkClient
List<TopicSink> sinks = getTopicEndpointManager().getTopicSinks(sinkTopic);
this.sink = sinks.get(0);
// init source
- List<TopicSource> sources = getTopicEndpointManager().getTopicSources(Arrays.asList(sourceTopic));
+ List<TopicSource> sources = getTopicEndpointManager().getTopicSources(Collections.singletonList(sourceTopic));
if (sources.isEmpty()) {
throw new BidirectionalTopicClientException("no sources for topic: " + sourceTopic);
} else if (sources.size() > 1) {
}
/**
- * Determines whether or not the topic is ready (i.e., {@link #awaitReady(Object)} has
+ * Determines whether the topic is ready (i.e., {@link #awaitReady(Object, long)} has
* previously returned {@code true}).
*
* @return {@code true}, if the topic is ready to send and receive
* Waits for the bidirectional topic to become "ready" by publishing a message on the
* sink topic and awaiting receipt of the message on the source topic. If the message
* is not received within a few seconds, then it tries again. This process is
- * continued until the message is received, {@link #stop()} is called, or this thread
+ * continued until the message is received, {@link #stopWaiting()} is called, or this thread
* is interrupted. Once this returns, subsequent calls will return immediately, always
* with the same value.
*
final String messageText = coder.encode(message);
// class of message to be decoded
- @SuppressWarnings("unchecked")
- final Class<? extends T> clazz = (Class<? extends T>) message.getClass();
-
- // create a listener to detect when a matching message is received
- final TopicListener listener = (infra, topic, msg) -> {
- try {
- T incoming = decode(msg, clazz);
-
- if (message.equals(incoming)) {
- logger.info("topic {} is ready; found matching message {}", topic, incoming);
- checkerQueue.add(Boolean.TRUE);
- }
-
- } catch (CoderException e) {
- logger.warn("cannot decode message from topic {}", topic, e);
- decodeFailed();
- }
- };
+ final TopicListener listener = getTopicListener(message);
source.register(listener);
return checkerQueue.peek();
}
+ @NotNull
+ private <T> TopicListener getTopicListener(T message) {
+ @SuppressWarnings("unchecked")
+ final Class<? extends T> clazz = (Class<? extends T>) message.getClass();
+
+ // create a listener to detect when a matching message is received
+ return (infra, topic, msg) -> {
+ try {
+ T incoming = decode(msg, clazz);
+
+ if (message.equals(incoming)) {
+ logger.info("topic {} is ready; found matching message {}", topic, incoming);
+ checkerQueue.add(Boolean.TRUE);
+ }
+
+ } catch (CoderException e) {
+ logger.warn("cannot decode message from topic {}", topic, e);
+ decodeFailed();
+ }
+ };
+ }
+
/**
* Stops any listeners that are currently stuck in {@link #awaitReady(Object)} by
* adding {@code false} to the queue.
* ONAP
* ================================================================================
* Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
package org.onap.policy.common.endpoints.event.comm.client;
+import java.io.Serial;
+
/**
* Exception thrown by BidirectionalTopicClient class.
*/
public class BidirectionalTopicClientException extends Exception {
+ @Serial
private static final long serialVersionUID = 1L;
public BidirectionalTopicClientException() {
* @throws TopicSinkClientException if the topic does not exist
*/
public TopicSinkClient(final String topic) throws TopicSinkClientException {
- final List<TopicSink> lst = getTopicSinks(topic);
+ final List<TopicSink> lst = getTopicSinks(topic.toLowerCase());
if (lst.isEmpty()) {
- throw new TopicSinkClientException("no sinks for topic: " + topic);
+ throw new TopicSinkClientException("no sinks for topic: " + topic.toLowerCase());
}
this.sink = lst.get(0);
* ONAP PAP
* ================================================================================
* Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
- * Modifications Copyright (C) 2019 Nordix Foundation.
+ * Modifications Copyright (C) 2019, 2023 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
package org.onap.policy.common.endpoints.event.comm.client;
+import java.io.Serial;
+
/**
* Exception thrown by TopicSink client classes.
*/
public class TopicSinkClientException extends Exception {
+ @Serial
private static final long serialVersionUID = 1L;
public TopicSinkClientException() {
Code Review / policy / common.git / commitdiff