Trigger pipeconf deploy right after registration

Without waiting for the next pipeconf watchdog periodic probe.
To support this, this patch extends the PiPipeconfService to advertise
pipeconf registration events.

Change-Id: Ib44f1813bd37083c666a5e7980de320ce469c2d2

apps/p4-tutorial/pipeconf/src/main/java/org/onosproject/p4tutorial/pipeconf/PipeconfFactory.java

@@ -70,7 +70,7 @@ public final class PipeconfFactory {
     public void deactivate() {
         // Unregisters the pipeconf at component deactivation.
         try {
-            piPipeconfService.remove(PIPECONF_ID);
+            piPipeconfService.unregister(PIPECONF_ID);
         } catch (IllegalStateException e) {
             log.warn("{} haven't been registered", PIPECONF_ID);
         }

core/api/src/main/java/org/onosproject/net/pi/service/PiPipeconfEvent.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * 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.onosproject.net.pi.service;
+
+import org.onosproject.event.AbstractEvent;
+import org.onosproject.net.pi.model.PiPipeconf;
+import org.onosproject.net.pi.model.PiPipeconfId;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Event related to the PiPipeconfService.
+ */
+public class PiPipeconfEvent extends AbstractEvent<PiPipeconfEvent.Type, PiPipeconfId> {
+
+    private final PiPipeconf pipeconf;
+
+    /**
+     * Type of pipeconf event.
+     */
+    public enum Type {
+        REGISTERED,
+        UNREGISTERED
+    }
+
+    /**
+     * Creates anew pipeconf event for the given type and pipeconf.
+     *
+     * @param type     type of event
+     * @param pipeconf pipeconf
+     */
+    public PiPipeconfEvent(Type type, PiPipeconf pipeconf) {
+        super(type, checkNotNull(pipeconf).id());
+        this.pipeconf = pipeconf;
+    }
+
+
+    /**
+     * Creates anew pipeconf event for the given type and pipeconf ID.
+     *
+     * @param type       type of event
+     * @param pipeconfId pipeconf ID
+     */
+    public PiPipeconfEvent(Type type, PiPipeconfId pipeconfId) {
+        super(type, pipeconfId);
+        pipeconf = null;
+    }
+
+    /**
+     * Returns the pipeconf instance associated to this event, or null if one
+     * was not provided. For example, {@link Type#UNREGISTERED} events are not
+     * expected to carry the pipeconf instance that was unregistered, but just
+     * the ID (via {@link #subject()}).
+     *
+     * @return pipeconf instance or null
+     */
+    public PiPipeconf pipeconf() {
+        return pipeconf;
+    }
+}

core/api/src/main/java/org/onosproject/net/pi/service/PiPipeconfListener.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * 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.onosproject.net.pi.service;
+
+import org.onosproject.event.EventListener;
+
+/**
+ * Listener of pipeconf events.
+ */
+public interface PiPipeconfListener extends EventListener<PiPipeconfEvent> {
+}

core/api/src/main/java/org/onosproject/net/pi/service/PiPipeconfService.java

@@ -17,6 +17,7 @@
 package org.onosproject.net.pi.service;
 
 import com.google.common.annotations.Beta;
+import org.onosproject.event.ListenerService;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.model.PiPipeconfId;
@@ -27,12 +28,12 @@ import java.util.Optional;
  * A service to manage the configurations of protocol-independent pipelines.
  */
 @Beta
-public interface PiPipeconfService {
+public interface PiPipeconfService extends ListenerService<PiPipeconfEvent, PiPipeconfListener> {
 
     // TODO: we might want to extend ListenerService to support the broadcasting of PipeconfEvent.
 
     /**
-     * Registers the given pipeconf.
+     * Registers the given pipeconf making it available to other subsystems.
      *
      * @param pipeconf a pipeconf
      * @throws IllegalStateException if the same pipeconf identifier is already
@@ -41,17 +42,14 @@ public interface PiPipeconfService {
     void register(PiPipeconf pipeconf) throws IllegalStateException;
 
     /**
-     * Unregisters the Pipeconf identified by the given PiPipeconfId.
+     * Unregisters the given pipeconf. Once unregistered, other subsystems will
-     * Unregistering a Pipeconf removes it from the ONOS controller, thus making
+     * not be able to access the pipeconf content.
-     * it un-capable of controlling (e.g installing flow rules) the devices that
-     * have the pipeconf's P4 program deployed. For now this method DOES NOT
-     * remove the P4 program from the devices.
      *
      * @param pipeconfId a pipeconfId
      * @throws IllegalStateException if the same pipeconf identifier is already
      *                               registered.
      */
-    void remove(PiPipeconfId pipeconfId) throws IllegalStateException;
+    void unregister(PiPipeconfId pipeconfId) throws IllegalStateException;
 
     /**
      * Returns all pipeconfs registered.

core/api/src/test/java/org/onosproject/net/pi/PiPipeconfServiceAdapter.java

@@ -19,6 +19,7 @@ package org.onosproject.net.pi;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.model.PiPipeconfId;
+import org.onosproject.net.pi.service.PiPipeconfListener;
 import org.onosproject.net.pi.service.PiPipeconfService;
 
 import java.util.Collections;
@@ -34,7 +35,7 @@ public class PiPipeconfServiceAdapter implements PiPipeconfService {
     }
 
     @Override
-    public void remove(PiPipeconfId pipeconfId) throws IllegalStateException {
+    public void unregister(PiPipeconfId pipeconfId) throws IllegalStateException {
 
     }
 
@@ -67,4 +68,14 @@ public class PiPipeconfServiceAdapter implements PiPipeconfService {
     public Optional<PiPipeconfId> ofDevice(DeviceId deviceId) {
         return Optional.empty();
     }
+
+    @Override
+    public void addListener(PiPipeconfListener listener) {
+
+    }
+
+    @Override
+    public void removeListener(PiPipeconfListener listener) {
+
+    }
 }

core/net/src/main/java/org/onosproject/net/pi/impl/PiPipeconfManager.java

@@ -23,6 +23,7 @@ import com.google.common.util.concurrent.Striped;
 import org.onlab.util.HexString;
 import org.onlab.util.ItemNotFoundException;
 import org.onlab.util.SharedExecutors;
+import org.onosproject.event.AbstractListenerManager;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.config.NetworkConfigRegistry;
 import org.onosproject.net.config.basics.BasicDeviceConfig;
@@ -36,6 +37,8 @@ import org.onosproject.net.driver.DriverListener;
 import org.onosproject.net.driver.DriverProvider;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.model.PiPipeconfId;
+import org.onosproject.net.pi.service.PiPipeconfEvent;
+import org.onosproject.net.pi.service.PiPipeconfListener;
 import org.onosproject.net.pi.service.PiPipeconfMappingStore;
 import org.onosproject.net.pi.service.PiPipeconfService;
 import org.osgi.service.component.annotations.Activate;
@@ -68,7 +71,9 @@ import static org.slf4j.LoggerFactory.getLogger;
  */
 @Component(immediate = true, service = PiPipeconfService.class)
 @Beta
-public class PiPipeconfManager implements PiPipeconfService {
+public class PiPipeconfManager
+        extends AbstractListenerManager<PiPipeconfEvent, PiPipeconfListener>
+        implements PiPipeconfService {
 
     private final Logger log = getLogger(getClass());
 
@@ -100,6 +105,7 @@ public class PiPipeconfManager implements PiPipeconfService {
     @Activate
     public void activate() {
         driverAdminService.addListener(driverListener);
+        eventDispatcher.addSink(PiPipeconfEvent.class, listenerRegistry);
         checkMissingMergedDrivers();
         if (!missingMergedDrivers.isEmpty()) {
             // Missing drivers should be created upon detecting registration
@@ -114,6 +120,7 @@ public class PiPipeconfManager implements PiPipeconfService {
 
     @Deactivate
     public void deactivate() {
+        eventDispatcher.removeSink(PiPipeconfEvent.class);
         executor.shutdown();
         driverAdminService.removeListener(driverListener);
         pipeconfs.clear();
@@ -133,10 +140,11 @@ public class PiPipeconfManager implements PiPipeconfService {
         log.info("New pipeconf registered: {} (fingerprint={})",
                  pipeconf.id(), HexString.toHexString(pipeconf.fingerprint()));
         executor.execute(() -> attemptMergeAll(pipeconf.id()));
+        post(new PiPipeconfEvent(PiPipeconfEvent.Type.REGISTERED, pipeconf));
     }
 
     @Override
-    public void remove(PiPipeconfId pipeconfId) throws IllegalStateException {
+    public void unregister(PiPipeconfId pipeconfId) throws IllegalStateException {
         checkNotNull(pipeconfId);
         // TODO add mechanism to remove from device.
         if (!pipeconfs.containsKey(pipeconfId)) {
@@ -147,6 +155,7 @@ public class PiPipeconfManager implements PiPipeconfService {
         final PiPipeconf pipeconf = pipeconfs.remove(pipeconfId);
         log.info("Unregistered pipeconf: {} (fingerprint={})",
                  pipeconfId, HexString.toHexString(pipeconf.fingerprint()));
+        post(new PiPipeconfEvent(PiPipeconfEvent.Type.UNREGISTERED, pipeconfId));
     }
 
     @Override

core/net/src/main/java/org/onosproject/net/pi/impl/PiPipeconfWatchdogManager.java

@@ -35,6 +35,8 @@ import org.onosproject.net.device.DeviceListener;
 import org.onosproject.net.device.DeviceService;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.model.PiPipeconfId;
+import org.onosproject.net.pi.service.PiPipeconfEvent;
+import org.onosproject.net.pi.service.PiPipeconfListener;
 import org.onosproject.net.pi.service.PiPipeconfMappingStore;
 import org.onosproject.net.pi.service.PiPipeconfService;
 import org.onosproject.net.pi.service.PiPipeconfWatchdogEvent;
@@ -57,6 +59,7 @@ import org.slf4j.Logger;
 
 import java.util.Dictionary;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.ExecutorService;
@@ -75,11 +78,11 @@ import static org.slf4j.LoggerFactory.getLogger;
  * pipeline.
  */
 @Component(
-    immediate = true,
+        immediate = true,
-    service = PiPipeconfWatchdogService.class,
+        service = PiPipeconfWatchdogService.class,
-    property = {
+        property = {
-        PWM_PROBE_INTERVAL + ":Integer=" + PWM_PROBE_INTERVAL_DEFAULT
+                PWM_PROBE_INTERVAL + ":Integer=" + PWM_PROBE_INTERVAL_DEFAULT
-    }
+        }
 )
 public class PiPipeconfWatchdogManager
         extends AbstractListenerManager<PiPipeconfWatchdogEvent, PiPipeconfWatchdogListener>
@@ -107,13 +110,16 @@ public class PiPipeconfWatchdogManager
     @Reference(cardinality = ReferenceCardinality.MANDATORY)
     private ComponentConfigService componentConfigService;
 
-    /** Configure interval in seconds for device pipeconf probing. */
+    /**
+     * Configure interval in seconds for device pipeconf probing.
+     */
     private int probeInterval = PWM_PROBE_INTERVAL_DEFAULT;
 
     protected ExecutorService executor = Executors.newFixedThreadPool(
             30, groupedThreads("onos/pipeconf-watchdog", "%d", log));
 
-    private final InternalDeviceListener deviceListener = new InternalDeviceListener();
+    private final DeviceListener deviceListener = new InternalDeviceListener();
+    private final PiPipeconfListener pipeconfListener = new InternalPipeconfListener();
 
     private Timer timer;
     private TimerTask task;
@@ -141,8 +147,9 @@ public class PiPipeconfWatchdogManager
         // Start periodic watchdog task.
         timer = new Timer();
         startProbeTask();
-        // Add device listener.
+        // Add listeners.
         deviceService.addListener(deviceListener);
+        pipeconfService.addListener(pipeconfListener);
         log.info("Started");
     }
 
@@ -167,6 +174,7 @@ public class PiPipeconfWatchdogManager
     @Deactivate
     public void deactivate() {
         eventDispatcher.removeSink(PiPipeconfWatchdogEvent.class);
+        pipeconfService.removeListener(pipeconfListener);
         deviceService.removeListener(deviceListener);
         stopProbeTask();
         timer = null;
@@ -205,7 +213,8 @@ public class PiPipeconfWatchdogManager
             }
 
             if (!pipeconfService.getPipeconf(pipeconfId).isPresent()) {
-                log.error("Pipeconf {} is not registered", pipeconfId);
+                log.warn("Pipeconf {} is not registered, skipping probe for {}",
+                         pipeconfId, device.id());
                 return;
             }
 
@@ -356,6 +365,19 @@ public class PiPipeconfWatchdogManager
         }
     }
 
+    private class InternalPipeconfListener implements PiPipeconfListener {
+        @Override
+        public void event(PiPipeconfEvent event) {
+            pipeconfMappingStore.getDevices(event.subject())
+                    .forEach(PiPipeconfWatchdogManager.this::triggerProbe);
+        }
+
+        @Override
+        public boolean isRelevant(PiPipeconfEvent event) {
+            return Objects.equals(event.type(), PiPipeconfEvent.Type.REGISTERED);
+        }
+    }
+
     private class StatusMapListener
             implements EventuallyConsistentMapListener<DeviceId, PipelineStatus> {
 

core/net/src/test/java/org/onosproject/net/pi/impl/PiPipeconfManagerTest.java

@@ -24,6 +24,7 @@ import com.google.common.collect.Sets;
 import org.junit.Before;
 import org.junit.Test;
 import org.onlab.util.ItemNotFoundException;
+import org.onosproject.common.event.impl.TestEventDispatcher;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.config.Config;
 import org.onosproject.net.config.ConfigApplyDelegate;
@@ -55,7 +56,9 @@ import java.util.Map;
 import java.util.Set;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
+import static org.onosproject.net.NetTestTools.injectEventDispatcher;
 import static org.onosproject.pipelines.basic.PipeconfLoader.BASIC_PIPECONF;
 
 
@@ -82,49 +85,50 @@ public class PiPipeconfManagerTest {
 
 
     //Services
-    private PiPipeconfManager piPipeconfService;
+    private PiPipeconfManager mgr;
-    private PiPipeconf piPipeconf;
+    private PiPipeconf pipeconf;
 
     @Before
     public void setUp() throws IOException {
-        piPipeconfService = new PiPipeconfManager();
+        mgr = new PiPipeconfManager();
-        piPipeconf = BASIC_PIPECONF;
+        pipeconf = BASIC_PIPECONF;
-        piPipeconfService.cfgService = cfgService;
+        mgr.cfgService = cfgService;
-        piPipeconfService.driverAdminService = driverAdminService;
+        mgr.driverAdminService = driverAdminService;
+        injectEventDispatcher(mgr, new TestEventDispatcher());
         ObjectMapper mapper = new ObjectMapper();
         ConfigApplyDelegate delegate = new MockDelegate();
         String keyBasic = "basic";
         JsonNode jsonNodeBasic = mapper.readTree(jsonStreamBasic);
         basicDeviceConfig.init(DEVICE_ID, keyBasic, jsonNodeBasic, mapper, delegate);
-        piPipeconfService.activate();
+        mgr.activate();
     }
 
     @Test
     public void activate() {
-        assertEquals("Incorrect driver admin service", driverAdminService, piPipeconfService.driverAdminService);
+        assertEquals("Incorrect driver admin service", driverAdminService, mgr.driverAdminService);
-        assertEquals("Incorrect driverAdminService service", driverAdminService, piPipeconfService.driverAdminService);
+        assertEquals("Incorrect driverAdminService service", driverAdminService, mgr.driverAdminService);
-        assertEquals("Incorrect configuration service", cfgService, piPipeconfService.cfgService);
+        assertEquals("Incorrect configuration service", cfgService, mgr.cfgService);
     }
 
     @Test
     public void deactivate() {
-        piPipeconfService.deactivate();
+        mgr.deactivate();
-        assertEquals("Incorrect driver admin service", null, piPipeconfService.driverAdminService);
+        assertNull("Incorrect driver admin service", mgr.driverAdminService);
-        assertEquals("Incorrect driverAdminService service", null, piPipeconfService.driverAdminService);
+        assertNull("Incorrect driverAdminService service", mgr.driverAdminService);
-        assertEquals("Incorrect configuration service", null, piPipeconfService.cfgService);
+        assertNull("Incorrect configuration service", mgr.cfgService);
     }
 
     @Test
     public void register() {
-        piPipeconfService.register(piPipeconf);
+        mgr.register(pipeconf);
-        assertTrue("PiPipeconf should be registered", piPipeconfService.pipeconfs.containsValue(piPipeconf));
+        assertTrue("PiPipeconf should be registered", mgr.pipeconfs.containsValue(pipeconf));
     }
 
     @Test
     public void getPipeconf() {
-        piPipeconfService.register(piPipeconf);
+        mgr.register(pipeconf);
-        assertEquals("Returned PiPipeconf is not correct", piPipeconf,
+        assertEquals("Returned PiPipeconf is not correct", pipeconf,
-                     piPipeconfService.getPipeconf(piPipeconf.id()).get());
+                     mgr.getPipeconf(pipeconf.id()).get());
     }
 
 
@@ -132,29 +136,29 @@ public class PiPipeconfManagerTest {
     public void mergeDriver() {
         PiPipeconfId piPipeconfId = new PiPipeconfId(cfgService.getConfig(
                 DEVICE_ID, BasicDeviceConfig.class).pipeconf());
-        assertEquals(piPipeconf.id(), piPipeconfId);
+        assertEquals(pipeconf.id(), piPipeconfId);
 
         String baseDriverName = cfgService.getConfig(DEVICE_ID, BasicDeviceConfig.class).driver();
         assertEquals(BASE_DRIVER, baseDriverName);
 
-        piPipeconfService.register(piPipeconf);
+        mgr.register(pipeconf);
-        assertEquals("Returned PiPipeconf is not correct", piPipeconf,
+        assertEquals("Returned PiPipeconf is not correct", pipeconf,
-                     piPipeconfService.getPipeconf(piPipeconf.id()).get());
+                     mgr.getPipeconf(pipeconf.id()).get());
 
-        String mergedDriverName = piPipeconfService.getMergedDriver(DEVICE_ID, piPipeconfId);
+        String mergedDriverName = mgr.getMergedDriver(DEVICE_ID, piPipeconfId);
 
         String expectedName = BASE_DRIVER + ":" + piPipeconfId.id();
         assertEquals(expectedName, mergedDriverName);
 
         //we assume that the provider is 1 and that it contains 1 driver
         //we also assume that everything after driverAdminService.registerProvider(provider); has been tested.
-        assertTrue("Provider should be registered", providers.size() == 1);
+        assertEquals("Provider should be registered", 1, providers.size());
 
         assertTrue("Merged driver name should be valid",
                    mergedDriverName != null && !mergedDriverName.isEmpty());
 
         DriverProvider provider = providers.iterator().next();
-        assertTrue("Provider should contain one driver", provider.getDrivers().size() == 1);
+        assertEquals("Provider should contain one driver", 1, provider.getDrivers().size());
 
         Driver driver = provider.getDrivers().iterator().next();
 

core/store/dist/src/main/java/org/onosproject/store/pi/impl/DistributedDevicePipeconfMappingStore.java

@@ -58,7 +58,7 @@ public class DistributedDevicePipeconfMappingStore
 
     protected ConsistentMap<DeviceId, PiPipeconfId> deviceToPipeconf;
 
-    protected final MapEventListener<DeviceId, PiPipeconfId> pipeconfListener =
+    protected final MapEventListener<DeviceId, PiPipeconfId> mapListener =
             new InternalPiPipeconfListener();
 
     protected SetMultimap<PiPipeconfId, DeviceId> pipeconfToDevices =
@@ -70,13 +70,13 @@ public class DistributedDevicePipeconfMappingStore
                 .withName("onos-pipeconf-table")
                 .withSerializer(Serializer.using(KryoNamespaces.API))
                 .build();
-        deviceToPipeconf.addListener(pipeconfListener);
+        deviceToPipeconf.addListener(mapListener);
         log.info("Started");
     }
 
     @Deactivate
     public void deactivate() {
-        deviceToPipeconf.removeListener(pipeconfListener);
+        deviceToPipeconf.removeListener(mapListener);
         deviceToPipeconf = null;
         pipeconfToDevices = null;
         log.info("Stopped");

pipelines/basic/src/main/java/org/onosproject/pipelines/basic/PipeconfLoader.java

@@ -77,7 +77,7 @@ public final class PipeconfLoader {
 
     @Deactivate
     public void deactivate() {
-        ALL_PIPECONFS.stream().map(PiPipeconf::id).forEach(piPipeconfService::remove);
+        ALL_PIPECONFS.stream().map(PiPipeconf::id).forEach(piPipeconfService::unregister);
     }
 
     private static PiPipeconf buildBasicPipeconf() {

pipelines/fabric/src/main/java/org/onosproject/pipelines/fabric/PipeconfLoader.java

@@ -101,7 +101,7 @@ public class PipeconfLoader {
 
     @Deactivate
     public void deactivate() {
-        PIPECONFS.stream().map(PiPipeconf::id).forEach(piPipeconfService::remove);
+        PIPECONFS.stream().map(PiPipeconf::id).forEach(piPipeconfService::unregister);
     }
 
     private static Collection<PiPipeconf> buildAllPipeconf() {

protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/client/WriteRequestImpl.java

@@ -18,6 +18,7 @@ package org.onosproject.p4runtime.ctl.client;
 
 import com.google.common.util.concurrent.Futures;
 import com.google.protobuf.TextFormat;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.runtime.PiEntity;
@@ -35,6 +36,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 import static java.lang.String.format;
 import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.onosproject.p4runtime.api.P4RuntimeWriteClient.EntityUpdateStatus.PENDING;
 import static org.onosproject.p4runtime.ctl.client.P4RuntimeClientImpl.SHORT_TIMEOUT_SECONDS;
 import static org.onosproject.p4runtime.ctl.codec.Codecs.CODECS;
 import static org.slf4j.LoggerFactory.getLogger;
@@ -190,8 +192,18 @@ final class WriteRequestImpl implements P4RuntimeWriteClient.WriteRequest {
 
                     @Override
                     public void onError(Throwable t) {
-                        client.handleRpcError(t, "WRITE");
+                        final WriteResponseImpl response = responseBuilder
-                        future.complete(responseBuilder.setErrorsAndBuild(t));
+                                .setErrorsAndBuild(t);
+                        if (Status.fromThrowable(t).getCode() != Status.Code.UNKNOWN
+                                || !response.status(PENDING).isEmpty()) {
+                            // If UNKNOWN and no entities are in PENDING state,
+                            // it means we have processed the response error
+                            // details and a log message will be produced for
+                            // each failed entity. No need to log the top level
+                            // SRE.
+                            client.handleRpcError(t, "WRITE");
+                        }
+                        future.complete(response);
                     }
 
                     @Override

protocols/p4runtime/ctl/src/test/java/org/onosproject/p4runtime/ctl/MockPipeconfService.java

@@ -19,6 +19,7 @@ package org.onosproject.p4runtime.ctl;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.pi.model.PiPipeconf;
 import org.onosproject.net.pi.model.PiPipeconfId;
+import org.onosproject.net.pi.service.PiPipeconfListener;
 import org.onosproject.net.pi.service.PiPipeconfService;
 
 import java.util.Optional;
@@ -30,7 +31,7 @@ public class MockPipeconfService implements PiPipeconfService {
     }
 
     @Override
-    public void remove(PiPipeconfId pipeconfId) throws IllegalStateException {
+    public void unregister(PiPipeconfId pipeconfId) throws IllegalStateException {
 
     }
 
@@ -63,4 +64,14 @@ public class MockPipeconfService implements PiPipeconfService {
     public Optional<PiPipeconfId> ofDevice(DeviceId deviceId) {
         return Optional.empty();
     }
+
+    @Override
+    public void addListener(PiPipeconfListener listener) {
+
+    }
+
+    @Override
+    public void removeListener(PiPipeconfListener listener) {
+
+    }
 }