diff --git a/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/AbstractBolt.java b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/AbstractBolt.java index fb02cde4a1d..d7f857e5353 100644 --- a/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/AbstractBolt.java +++ b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/AbstractBolt.java @@ -66,10 +66,10 @@ public abstract class AbstractBolt extends BaseRichBolt { @Getter(AccessLevel.PROTECTED) private transient Integer taskId; - @Getter(AccessLevel.PROTECTED) + @Getter(AccessLevel.PUBLIC) private transient Tuple currentTuple; - @Getter(AccessLevel.PROTECTED) + @Getter(AccessLevel.PUBLIC) @Setter(AccessLevel.PROTECTED) private transient CommandContext commandContext; @@ -137,7 +137,14 @@ protected void emit(String stream, List payload) { output.emit(stream, payload); } - protected void emitWithContext(String stream, Tuple input, Values payload) { + /** + * Emitting (sending a message) the given payload with the anchor based on the tuple to the given stream. + * This guarantees that the message will be eventually processed by Storm even in cases of the target bolt failure. + * @param stream Storm stream to send the message to + * @param input current tuple; will be used as an anchor + * @param payload the message payload + */ + public void emitWithContext(String stream, Tuple input, Values payload) { payload.add(getCommandContext()); log.debug("emit tuple into {} stream: {}", stream, payload); getOutput().emit(stream, input, payload); diff --git a/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/ContextEmittable.java b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/ContextEmittable.java new file mode 100644 index 00000000000..b6b0f6b028d --- /dev/null +++ b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/ContextEmittable.java @@ -0,0 +1,29 @@ +/* Copyright 2023 Telstra Open Source + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.openkilda.wfm; + +import org.apache.storm.tuple.Tuple; +import org.apache.storm.tuple.Values; + +/** + * This interface represents something that has CommandContext and + * knows how to emit the payload via the given Storm stream. + */ +public interface ContextEmittable { + CommandContext getCommandContext(); + + void emitWithContext(String stream, Tuple input, Values payload); +} diff --git a/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/DefaultHistoryUpdateCarrier.java b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/DefaultHistoryUpdateCarrier.java new file mode 100644 index 00000000000..5f2f5270fe5 --- /dev/null +++ b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/DefaultHistoryUpdateCarrier.java @@ -0,0 +1,40 @@ +/* Copyright 2023 Telstra Open Source + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.openkilda.wfm; + +import org.openkilda.messaging.info.InfoMessage; +import org.openkilda.wfm.share.history.model.FlowHistoryHolder; + +import lombok.NonNull; +import org.apache.storm.tuple.Values; + +/** + * This is a default implementation for sending history updates that are accessible later via HistoryService. + * A topology that uses the bolt implementing this interface must register a path to history bolt. + * That is, in a topology that extends AbstractTopology, declare a Kafka bolt and specify a component ID + * that is a source of history updates in the fields grouping and declare a stream. + * TODO add a readme.md with the details and example. + */ +public interface DefaultHistoryUpdateCarrier extends HistoryUpdateCarrier, TupleIdentifiable, ContextEmittable { + + @Override + default void sendHistoryUpdate(@NonNull FlowHistoryHolder historyHolder) { + InfoMessage message = new InfoMessage(historyHolder, getCommandContext().getCreateTime(), + getCommandContext().getCorrelationId()); + emitWithContext(getHistoryStreamName(), getCurrentTuple(), + new Values(historyHolder.getTaskId(), message)); + } +} diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/service/common/HistoryUpdateCarrier.java b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/HistoryUpdateCarrier.java similarity index 87% rename from src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/service/common/HistoryUpdateCarrier.java rename to src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/HistoryUpdateCarrier.java index fe744c0e4a4..6282197e6eb 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/service/common/HistoryUpdateCarrier.java +++ b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/HistoryUpdateCarrier.java @@ -13,7 +13,7 @@ * limitations under the License. */ -package org.openkilda.wfm.topology.flowhs.service.common; +package org.openkilda.wfm; import org.openkilda.wfm.share.history.model.FlowHistoryHolder; @@ -25,4 +25,8 @@ public interface HistoryUpdateCarrier { * Sends main events to history bolt. */ void sendHistoryUpdate(FlowHistoryHolder historyHolder); + + default String getHistoryStreamName() { + return "HUB_TO_HISTORY_TOPOLOGY_SENDER"; + } } diff --git a/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/TupleIdentifiable.java b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/TupleIdentifiable.java new file mode 100644 index 00000000000..300717ed481 --- /dev/null +++ b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/TupleIdentifiable.java @@ -0,0 +1,25 @@ +/* Copyright 2023 Telstra Open Source + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.openkilda.wfm; + +import org.apache.storm.tuple.Tuple; + +/** + * This interface represents something that has a distinct tuple that can be used as an anchor in Storm. + */ +public interface TupleIdentifiable { + Tuple getCurrentTuple(); +} diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/bolts/FlowCreateHubBolt.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/bolts/FlowCreateHubBolt.java index 990f43cbb01..fa247ee1480 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/bolts/FlowCreateHubBolt.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/bolts/FlowCreateHubBolt.java @@ -41,10 +41,10 @@ import org.openkilda.pce.PathComputerFactory; import org.openkilda.persistence.PersistenceManager; import org.openkilda.server42.control.messaging.flowrtt.ActivateFlowMonitoringInfoData; +import org.openkilda.wfm.DefaultHistoryUpdateCarrier; import org.openkilda.wfm.error.PipelineException; import org.openkilda.wfm.share.flow.resources.FlowResourcesConfig; import org.openkilda.wfm.share.flow.resources.FlowResourcesManager; -import org.openkilda.wfm.share.history.model.FlowHistoryHolder; import org.openkilda.wfm.share.hubandspoke.HubBolt; import org.openkilda.wfm.share.utils.KeyProvider; import org.openkilda.wfm.share.zk.ZkStreams; @@ -66,7 +66,7 @@ import org.apache.storm.tuple.Tuple; import org.apache.storm.tuple.Values; -public class FlowCreateHubBolt extends HubBolt implements FlowGenericCarrier { +public class FlowCreateHubBolt extends HubBolt implements FlowGenericCarrier, DefaultHistoryUpdateCarrier { private final FlowCreateConfig config; private final PathComputerConfig pathComputerConfig; private final FlowResourcesConfig flowResourcesConfig; @@ -161,14 +161,6 @@ public void sendNorthboundResponse(@NonNull Message message) { emitWithContext(Stream.HUB_TO_NB_RESPONSE_SENDER.name(), getCurrentTuple(), new Values(currentKey, message)); } - @Override - public void sendHistoryUpdate(@NonNull FlowHistoryHolder historyHolder) { - InfoMessage message = new InfoMessage(historyHolder, getCommandContext().getCreateTime(), - getCommandContext().getCorrelationId()); - emitWithContext(Stream.HUB_TO_HISTORY_TOPOLOGY_SENDER.name(), getCurrentTuple(), - new Values(historyHolder.getTaskId(), message)); - } - @Override public void cancelTimeoutCallback(@NonNull String key) { cancelCallback(key); diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowPathSwappingFsm.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowPathSwappingFsm.java index 319aacc341f..c81ec6bb70e 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowPathSwappingFsm.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowPathSwappingFsm.java @@ -26,10 +26,10 @@ import org.openkilda.model.PathId; import org.openkilda.model.SwitchId; import org.openkilda.wfm.CommandContext; +import org.openkilda.wfm.HistoryUpdateCarrier; import org.openkilda.wfm.share.flow.resources.FlowResources; import org.openkilda.wfm.topology.flowhs.model.RequestedFlow; import org.openkilda.wfm.topology.flowhs.service.FlowProcessingEventListener; -import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier; import org.openkilda.wfm.topology.flowhs.service.common.NorthboundResponseCarrier; import lombok.Getter; diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithHistorySupportFsm.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithHistorySupportFsm.java index ccad2ec143c..c7b3e786450 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithHistorySupportFsm.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithHistorySupportFsm.java @@ -18,13 +18,13 @@ import static java.util.Collections.emptyList; import org.openkilda.wfm.CommandContext; +import org.openkilda.wfm.HistoryUpdateCarrier; import org.openkilda.wfm.share.history.model.FlowDumpData; import org.openkilda.wfm.share.history.model.FlowEventData; import org.openkilda.wfm.share.history.model.FlowHistoryData; import org.openkilda.wfm.share.history.model.FlowHistoryHolder; import org.openkilda.wfm.share.utils.KeyProvider; import org.openkilda.wfm.topology.flowhs.service.common.FlowHistoryCarrier; -import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier; import org.openkilda.wfm.topology.flowhs.service.common.NorthboundResponseCarrier; import org.openkilda.wfm.topology.flowhs.service.common.ProcessingEventListener; diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithSpeakerCommandsFsm.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithSpeakerCommandsFsm.java index 3d778f8b77c..fe203e63060 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithSpeakerCommandsFsm.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithSpeakerCommandsFsm.java @@ -21,8 +21,8 @@ import org.openkilda.floodlight.api.response.rulemanager.SpeakerCommandResponse; import org.openkilda.model.SwitchId; import org.openkilda.wfm.CommandContext; +import org.openkilda.wfm.HistoryUpdateCarrier; import org.openkilda.wfm.topology.flowhs.service.FlowProcessingEventListener; -import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier; import org.openkilda.wfm.topology.flowhs.service.common.NorthboundResponseCarrier; import lombok.Getter; diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/HaFlowPathSwappingFsm.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/HaFlowPathSwappingFsm.java index 3daaf2638a4..642f87528b7 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/HaFlowPathSwappingFsm.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/HaFlowPathSwappingFsm.java @@ -31,12 +31,12 @@ import org.openkilda.pce.GetHaPathsResult; import org.openkilda.pce.HaPath; import org.openkilda.wfm.CommandContext; +import org.openkilda.wfm.HistoryUpdateCarrier; import org.openkilda.wfm.share.flow.resources.HaFlowResources; import org.openkilda.wfm.share.flow.resources.HaPathIdsPair; import org.openkilda.wfm.share.flow.resources.HaPathIdsPair.HaFlowPathIds; import org.openkilda.wfm.topology.flowhs.fsm.common.context.SpeakerResponseContext; import org.openkilda.wfm.topology.flowhs.service.FlowProcessingEventListener; -import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier; import org.openkilda.wfm.topology.flowhs.service.common.NorthboundResponseCarrier; import lombok.Getter; diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/haflow/BaseHaFlowPathRemovalAction.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/haflow/BaseHaFlowPathRemovalAction.java index 2c00ca2be8b..9231b93c8ab 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/haflow/BaseHaFlowPathRemovalAction.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/haflow/BaseHaFlowPathRemovalAction.java @@ -20,10 +20,10 @@ import org.openkilda.model.PathId; import org.openkilda.persistence.PersistenceManager; import org.openkilda.persistence.repositories.HaFlowPathRepository; +import org.openkilda.wfm.HistoryUpdateCarrier; import org.openkilda.wfm.share.flow.resources.HaPathIdsPair; import org.openkilda.wfm.topology.flowhs.fsm.common.FlowProcessingWithHistorySupportFsm; import org.openkilda.wfm.topology.flowhs.fsm.common.actions.BaseFlowPathRemovalAction; -import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier; import org.openkilda.wfm.topology.flowhs.service.haflow.history.HaFlowHistory; import org.openkilda.wfm.topology.flowhs.service.haflow.history.HaFlowHistoryService; diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/haflow/RevertResourceAllocationAction.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/haflow/RevertResourceAllocationAction.java index ab203884c48..c9dddf348fd 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/haflow/RevertResourceAllocationAction.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/haflow/RevertResourceAllocationAction.java @@ -19,11 +19,11 @@ import org.openkilda.model.HaFlow; import org.openkilda.persistence.PersistenceManager; +import org.openkilda.wfm.HistoryUpdateCarrier; import org.openkilda.wfm.share.flow.resources.FlowResourcesManager; import org.openkilda.wfm.share.flow.resources.HaFlowResources; import org.openkilda.wfm.topology.flowhs.fsm.common.HaFlowPathSwappingFsm; import org.openkilda.wfm.topology.flowhs.fsm.common.context.SpeakerResponseContext; -import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier; import org.openkilda.wfm.topology.flowhs.service.haflow.history.HaFlowHistory; import org.openkilda.wfm.topology.flowhs.service.haflow.history.HaFlowHistoryService; diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/haflow/reroute/actions/CompleteFlowPathRemovalAction.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/haflow/reroute/actions/CompleteFlowPathRemovalAction.java index b24e1c582fc..71afce67fa8 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/haflow/reroute/actions/CompleteFlowPathRemovalAction.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/haflow/reroute/actions/CompleteFlowPathRemovalAction.java @@ -16,13 +16,13 @@ package org.openkilda.wfm.topology.flowhs.fsm.haflow.reroute.actions; import org.openkilda.persistence.PersistenceManager; +import org.openkilda.wfm.HistoryUpdateCarrier; import org.openkilda.wfm.share.metrics.TimedExecution; import org.openkilda.wfm.topology.flowhs.fsm.common.actions.haflow.BaseHaFlowPathRemovalAction; import org.openkilda.wfm.topology.flowhs.fsm.haflow.reroute.HaFlowRerouteContext; import org.openkilda.wfm.topology.flowhs.fsm.haflow.reroute.HaFlowRerouteFsm; import org.openkilda.wfm.topology.flowhs.fsm.haflow.reroute.HaFlowRerouteFsm.Event; import org.openkilda.wfm.topology.flowhs.fsm.haflow.reroute.HaFlowRerouteFsm.State; -import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier; import lombok.extern.slf4j.Slf4j; diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/haflow/update/actions/CompleteFlowPathRemovalAction.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/haflow/update/actions/CompleteFlowPathRemovalAction.java index 5758a14c338..1bb640f0db8 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/haflow/update/actions/CompleteFlowPathRemovalAction.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/haflow/update/actions/CompleteFlowPathRemovalAction.java @@ -16,12 +16,12 @@ package org.openkilda.wfm.topology.flowhs.fsm.haflow.update.actions; import org.openkilda.persistence.PersistenceManager; +import org.openkilda.wfm.HistoryUpdateCarrier; import org.openkilda.wfm.topology.flowhs.fsm.common.actions.haflow.BaseHaFlowPathRemovalAction; import org.openkilda.wfm.topology.flowhs.fsm.haflow.update.HaFlowUpdateContext; import org.openkilda.wfm.topology.flowhs.fsm.haflow.update.HaFlowUpdateFsm; import org.openkilda.wfm.topology.flowhs.fsm.haflow.update.HaFlowUpdateFsm.Event; import org.openkilda.wfm.topology.flowhs.fsm.haflow.update.HaFlowUpdateFsm.State; -import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier; import lombok.extern.slf4j.Slf4j; diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/service/FlowGenericCarrier.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/service/FlowGenericCarrier.java index 6d72ddc30f1..f1647d8e9e4 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/service/FlowGenericCarrier.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/service/FlowGenericCarrier.java @@ -21,8 +21,8 @@ import org.openkilda.messaging.info.stats.StatsNotification; import org.openkilda.messaging.info.stats.UpdateFlowPathInfo; import org.openkilda.model.SwitchId; +import org.openkilda.wfm.HistoryUpdateCarrier; import org.openkilda.wfm.topology.flowhs.model.RequestedFlow; -import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier; import org.openkilda.wfm.topology.flowhs.service.common.LifecycleEventCarrier; import org.openkilda.wfm.topology.flowhs.service.common.NorthboundResponseCarrier; diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/service/haflow/history/HaFlowHistoryService.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/service/haflow/history/HaFlowHistoryService.java index 82e7eadf3ea..8cb8c2bdb52 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/service/haflow/history/HaFlowHistoryService.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/service/haflow/history/HaFlowHistoryService.java @@ -15,10 +15,10 @@ package org.openkilda.wfm.topology.flowhs.service.haflow.history; +import org.openkilda.wfm.HistoryUpdateCarrier; import org.openkilda.wfm.share.history.model.FlowHistoryHolder; import org.openkilda.wfm.share.history.model.HaFlowEventData; import org.openkilda.wfm.share.history.model.HaFlowHistoryData; -import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.ObjectUtils; diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/test/java/org/openkilda/wfm/topology/flowhs/service/haflow/history/HaFlowHistoryServiceTest.java b/src-java/flowhs-topology/flowhs-storm-topology/src/test/java/org/openkilda/wfm/topology/flowhs/service/haflow/history/HaFlowHistoryServiceTest.java index eb2e9851de5..38a2626bce9 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/test/java/org/openkilda/wfm/topology/flowhs/service/haflow/history/HaFlowHistoryServiceTest.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/test/java/org/openkilda/wfm/topology/flowhs/service/haflow/history/HaFlowHistoryServiceTest.java @@ -23,12 +23,12 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import org.openkilda.wfm.HistoryUpdateCarrier; import org.openkilda.wfm.share.history.model.FlowHistoryHolder; import org.openkilda.wfm.share.history.model.HaFlowDumpData; import org.openkilda.wfm.share.history.model.HaFlowEventData; import org.openkilda.wfm.share.history.model.HaFlowEventData.Event; import org.openkilda.wfm.share.history.model.HaFlowEventData.Initiator; -import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test;