class AddressObservationWriter {

    public void addAddress(MacAddress macAddress, IpAddress ipAddress, NodeConnectorRef nodeConnectorRef) {
        if (macAddress == null || ipAddress == null || nodeConnectorRef == null) {
            return;
        }

        // get the lock for given node connector so at a time only one
        // observation can be made on a node connector
        NodeConnectorLock nodeConnectorLock = lockMap.computeIfAbsent(nodeConnectorRef, key -> new NodeConnectorLock());

        synchronized (nodeConnectorLock) {
            // Ensure previous transaction finished writing to the db
            ListenableFuture<Void> future = futureMap.get(nodeConnectorLock);
            if (future != null) {
                try {
                    future.get();
                } catch (InterruptedException | ExecutionException e) {
                    LOG.error("Exception while waiting for previous transaction to finish", e);
                }
            }

            // Initialize builders
            long now = new Date().getTime();
            final AddressCapableNodeConnectorBuilder acncBuilder = new AddressCapableNodeConnectorBuilder();
            final AddressesBuilder addressBuilder = new AddressesBuilder().setIp(ipAddress).setMac(macAddress)
                    .setFirstSeen(now).setLastSeen(now);
            List<Addresses> addresses = null;

            // Read existing address observations from data tree
            ReadOnlyTransaction readTransaction = dataService.newReadOnlyTransaction();

            NodeConnector nc = null;
            try {
                Optional<NodeConnector> dataObjectOptional = readTransaction.read(LogicalDatastoreType.OPERATIONAL,
                        (InstanceIdentifier<NodeConnector>) nodeConnectorRef.getValue()).get();
                if (dataObjectOptional.isPresent()) {
                    nc = dataObjectOptional.get();
                }
            } catch (InterruptedException | ExecutionException e) {
                LOG.error("Error reading node connector {}", nodeConnectorRef.getValue());
                readTransaction.close();
                throw new RuntimeException("Error reading from operational store, node connector : " + nodeConnectorRef,
                        e);
            }
            readTransaction.close();
            if (nc == null) {
                return;
            }
            AddressCapableNodeConnector acnc = nc
                    .getAugmentation(AddressCapableNodeConnector.class);

            // Address observations exist
            if (acnc != null && acnc.getAddresses() != null) {
                // Search for this mac-ip pair in the existing address
                // observations & update last-seen timestamp
                addresses = acnc.getAddresses();
                for (int i = 0; i < addresses.size(); i++) {
                    if (addresses.get(i).getIp().equals(ipAddress) && addresses.get(i).getMac().equals(macAddress)) {
                        if (now - addresses.get(i).getLastSeen() > timestampUpdateInterval) {
                            addressBuilder.setFirstSeen(addresses.get(i).getFirstSeen())
                                    .setKey(addresses.get(i).getKey());
                            addresses.remove(i);
                            break;
                        } else {
                            return;
                        }
                    }
                }
            }
            // Address observations don't exist, so create the list
            else {
                addresses = new ArrayList<>();
            }

            if (addressBuilder.getKey() == null) {
                addressBuilder.setKey(new AddressesKey(BigInteger.valueOf(addressKey.getAndIncrement())));
            }

            // Add as an augmentation
            addresses.add(addressBuilder.build());
            acncBuilder.setAddresses(addresses);

            // build Instance Id for AddressCapableNodeConnector
            InstanceIdentifier<AddressCapableNodeConnector> addressCapableNcInstanceId =
                    ((InstanceIdentifier<NodeConnector>) nodeConnectorRef
                            .getValue()).augmentation(AddressCapableNodeConnector.class);
            final WriteTransaction writeTransaction = dataService.newWriteOnlyTransaction();
            // Update this AddressCapableNodeConnector in the MD-SAL data tree
            writeTransaction.merge(LogicalDatastoreType.OPERATIONAL, addressCapableNcInstanceId, acncBuilder.build());
            final ListenableFuture<Void> writeTxResultFuture = writeTransaction.submit();
            Futures.addCallback(writeTxResultFuture, new FutureCallback<Void>() {
                @Override
                public void onSuccess(Void notUsed) {
                    LOG.debug("AddressObservationWriter write successful for tx :{}",
                            writeTransaction.getIdentifier());
                }

                @Override
                public void onFailure(Throwable throwable) {
                    LOG.error("AddressObservationWriter write transaction {} failed",
                            writeTransaction.getIdentifier(), throwable.getCause());
                }
            }, MoreExecutors.directExecutor());
            futureMap.put(nodeConnectorLock, writeTxResultFuture);
        }
    }

    public  AddressObservationWriter(DataBroker dataService);

}