diff --git a/src/main/java/org/opencord/kafka/integrations/DhcpL2RelayKafkaIntegration.java b/src/main/java/org/opencord/kafka/integrations/DhcpL2RelayKafkaIntegration.java
index e1ca65a..f5f1636 100644
--- a/src/main/java/org/opencord/kafka/integrations/DhcpL2RelayKafkaIntegration.java
+++ b/src/main/java/org/opencord/kafka/integrations/DhcpL2RelayKafkaIntegration.java
@@ -19,12 +19,6 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
-import org.apache.felix.scr.annotations.Activate;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.Deactivate;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.ReferenceCardinality;
-import org.apache.felix.scr.annotations.ReferencePolicy;
 import org.onosproject.net.AnnotationKeys;
 import org.onosproject.net.device.DeviceService;
 import org.opencord.dhcpl2relay.DhcpAllocationInfo;
@@ -32,30 +26,34 @@
 import org.opencord.dhcpl2relay.DhcpL2RelayListener;
 import org.opencord.dhcpl2relay.DhcpL2RelayService;
 import org.opencord.kafka.EventBusService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+import org.osgi.service.component.annotations.ReferencePolicy;
 
 import java.time.Instant;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Listens for DHCP L2 relay events and pushes them on a Kafka bus.
  */
 @Component(immediate = true)
-public class DhcpL2RelayKafkaIntegration {
+public class DhcpL2RelayKafkaIntegration extends AbstractKafkaIntegration {
 
-    public Logger log = LoggerFactory.getLogger(getClass());
-
-    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
     protected EventBusService eventBusService;
 
-    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
     protected DeviceService deviceService;
 
-    @Reference(cardinality = ReferenceCardinality.OPTIONAL_UNARY,
+    @Reference(cardinality = ReferenceCardinality.OPTIONAL,
             policy = ReferencePolicy.DYNAMIC,
             bind = "bindDhcpL2RelayService",
             unbind = "unbindDhcpL2RelayService")
-    protected DhcpL2RelayService dhcpL2RelayService;
+    volatile protected DhcpL2RelayService ignore;
+    private final AtomicReference<DhcpL2RelayService> dhcpL2RelayServiceRef = new AtomicReference<>();
 
     private final DhcpL2RelayListener listener = new InternalDhcpL2RelayListener();
 
@@ -70,26 +68,12 @@
     private static final String MAC_ADDRESS = "macAddress";
     private static final String IP_ADDRESS = "ipAddress";
 
-    protected void bindDhcpL2RelayService(DhcpL2RelayService dhcpL2RelayService) {
-        if (this.dhcpL2RelayService == null) {
-            log.info("Binding DhcpL2RelayService");
-            this.dhcpL2RelayService = dhcpL2RelayService;
-            log.info("Adding listener on DhcpL2RelayService");
-            dhcpL2RelayService.addListener(listener);
-        } else {
-            log.warn("Trying to bind DhcpL2RelayService but it is already bound");
-        }
+    protected void bindDhcpL2RelayService(DhcpL2RelayService incomingService) {
+        bindAndAddListener(incomingService, dhcpL2RelayServiceRef, listener);
     }
 
-    protected void unbindDhcpL2RelayService(DhcpL2RelayService dhcpL2RelayService) {
-        if (this.dhcpL2RelayService == dhcpL2RelayService) {
-            log.info("Unbinding DhcpL2RelayService");
-            this.dhcpL2RelayService = null;
-            log.info("Removing listener on DhcpL2RelayService");
-            dhcpL2RelayService.removeListener(listener);
-        } else {
-            log.warn("Trying to unbind DhcpL2RelayService but it is already unbound");
-        }
+    protected void unbindDhcpL2RelayService(DhcpL2RelayService outgoingService) {
+        unbindAndRemoveListener(outgoingService, dhcpL2RelayServiceRef, listener);
     }
 
     @Activate
@@ -99,6 +83,7 @@
 
     @Deactivate
     public void deactivate() {
+        unbindDhcpL2RelayService(dhcpL2RelayServiceRef.get());
         log.info("Stopped DhcpL2RelayKafkaIntegration");
     }