diff --git a/adaptercore/openolt_flowmgr.go b/adaptercore/openolt_flowmgr.go
index 3376809..b8af453 100644
--- a/adaptercore/openolt_flowmgr.go
+++ b/adaptercore/openolt_flowmgr.go
@@ -28,16 +28,16 @@
 	"sync"
 	"time"
 
-	"github.com/opencord/voltha-lib-go/v2/pkg/flows"
-	"github.com/opencord/voltha-lib-go/v2/pkg/log"
-	tp "github.com/opencord/voltha-lib-go/v2/pkg/techprofile"
+	"github.com/opencord/voltha-lib-go/v3/pkg/flows"
+	"github.com/opencord/voltha-lib-go/v3/pkg/log"
+	tp "github.com/opencord/voltha-lib-go/v3/pkg/techprofile"
 	rsrcMgr "github.com/opencord/voltha-openolt-adapter/adaptercore/resourcemanager"
-	"github.com/opencord/voltha-protos/v2/go/common"
-	ic "github.com/opencord/voltha-protos/v2/go/inter_container"
-	ofp "github.com/opencord/voltha-protos/v2/go/openflow_13"
-	openoltpb2 "github.com/opencord/voltha-protos/v2/go/openolt"
-	tp_pb "github.com/opencord/voltha-protos/v2/go/tech_profile"
-	"github.com/opencord/voltha-protos/v2/go/voltha"
+	"github.com/opencord/voltha-protos/v3/go/common"
+	ic "github.com/opencord/voltha-protos/v3/go/inter_container"
+	ofp "github.com/opencord/voltha-protos/v3/go/openflow_13"
+	openoltpb2 "github.com/opencord/voltha-protos/v3/go/openolt"
+	tp_pb "github.com/opencord/voltha-protos/v3/go/tech_profile"
+	"github.com/opencord/voltha-protos/v3/go/voltha"
 
 	//deepcopy "github.com/getlantern/deepcopy"
 	"github.com/EagleChen/mapmutex"
@@ -57,6 +57,9 @@
 	//DhcpFlow flow category
 	DhcpFlow = "DHCP_FLOW"
 
+	//MulticastFlow flow category
+	MulticastFlow = "MULTICAST_FLOW"
+
 	//IgmpFlow flow category
 	IgmpFlow = "IGMP_FLOW"
 
@@ -90,6 +93,8 @@
 	Upstream = "upstream"
 	//Downstream constant
 	Downstream = "downstream"
+	//Multicast constant
+	Multicast = "multicast"
 	//PacketTagType constant
 	PacketTagType = "pkt_tag_type"
 	//Untagged constant
@@ -103,6 +108,8 @@
 
 	//EthType constant
 	EthType = "eth_type"
+	//EthDst constant
+	EthDst = "eth_dst"
 	//TPID constant
 	TPID = "tpid"
 	//IPProto constant
@@ -128,6 +135,8 @@
 	TunnelID = "tunnel_id"
 	//Output constant
 	Output = "output"
+	//GroupID constant
+	GroupID = "group_id"
 	// Actions
 
 	//PopVlan constant
@@ -152,6 +161,13 @@
 	PortNo = "portNo"
 	//AllocID constant
 	AllocID = "allocId"
+
+	//NoneOnuID constant
+	NoneOnuID = -1
+	//NoneUniID constant
+	NoneUniID = -1
+	//NoneGemPortID constant
+	NoneGemPortID = -1
 )
 
 type gemPortKey struct {
@@ -183,6 +199,11 @@
 	flowMetadata *voltha.FlowMetadata
 }
 
+type queueInfoBrief struct {
+	gemPortID       uint32
+	servicePriority uint32
+}
+
 //OpenOltFlowMgr creates the Structure of OpenOltFlowMgr obj
 type OpenOltFlowMgr struct {
 	techprofile        map[uint32]tp.TechProfileIf
@@ -195,7 +216,8 @@
 	lockCache          sync.RWMutex
 	pendingFlowDelete  sync.Map
 	// The mapmutex.Mutex can be fine tuned to use mapmutex.NewCustomizedMapMutex
-	perUserFlowHandleLock *mapmutex.Mutex
+	perUserFlowHandleLock    *mapmutex.Mutex
+	interfaceToMcastQueueMap map[uint32]*queueInfoBrief /*pon interface -> multicast queue map. Required to assign GEM to a bucket during group population*/
 }
 
 //NewFlowManager creates OpenOltFlowMgr object and initializes the parameters
@@ -228,6 +250,9 @@
 	flowMgr.lockCache = sync.RWMutex{}
 	flowMgr.pendingFlowDelete = sync.Map{}
 	flowMgr.perUserFlowHandleLock = mapmutex.NewMapMutex()
+	flowMgr.interfaceToMcastQueueMap = make(map[uint32]*queueInfoBrief)
+	//load interface to multicast queue map from kv store
+	flowMgr.loadInterfaceToMulticastQueueMap()
 	log.Info("Initialization of  flow manager success!!")
 	return &flowMgr
 }
@@ -239,6 +264,9 @@
 	} else if direction == Downstream {
 		log.Debug("downstream flow, not shifting id")
 		return uint64(flowID), nil
+	} else if direction == Multicast {
+		log.Debug("multicast flow, shifting id")
+		return 0x2<<15 | uint64(flowID), nil
 	} else {
 		log.Debug("Unrecognized direction")
 		return 0, fmt.Errorf("unrecognized direction %s", direction)
@@ -427,6 +455,25 @@
 		return err
 	}
 
+	if sq.direction == tp_pb.Direction_DOWNSTREAM {
+		multicastTrafficQueues := f.techprofile[sq.intfID].GetMulticastTrafficQueues(sq.tpInst)
+		if len(multicastTrafficQueues) > 0 {
+			if _, present := f.interfaceToMcastQueueMap[sq.intfID]; !present {
+				//assumed that there is only one queue per PON for the multicast service
+				//the default queue with multicastQueuePerPonPort.Priority per a pon interface is used for multicast service
+				//just put it in interfaceToMcastQueueMap to use for building group members
+				multicastQueuePerPonPort := multicastTrafficQueues[0]
+				f.interfaceToMcastQueueMap[sq.intfID] = &queueInfoBrief{
+					gemPortID:       multicastQueuePerPonPort.GemportId,
+					servicePriority: multicastQueuePerPonPort.Priority,
+				}
+				//also store the queue info in kv store
+				f.resourceMgr.AddMcastQueueForIntf(sq.intfID,
+					multicastQueuePerPonPort.GemportId,
+					multicastQueuePerPonPort.Priority)
+			}
+		}
+	}
 	return nil
 }
 
@@ -998,6 +1045,7 @@
 	classifier.DstPort, _ = classifierInfo[UDPDst].(uint32)
 	classifier.DstIp, _ = classifierInfo[Ipv4Dst].(uint32)
 	classifier.SrcIp, _ = classifierInfo[Ipv4Src].(uint32)
+	classifier.DstMac, _ = classifierInfo[EthDst].([]uint8)
 	if pktTagType, ok := classifierInfo[PacketTagType].(string); ok {
 		classifier.PktTagType = pktTagType
 
@@ -1505,6 +1553,12 @@
 func (f *OpenOltFlowMgr) clearFlowFromResourceManager(flow *ofp.OfpFlowStats, flowDirection string) {
 
 	log.Debugw("clearFlowFromResourceManager", log.Fields{"flowDirection": flowDirection, "flow": *flow})
+
+	if flowDirection == Multicast {
+		f.clearMulticastFlowFromResourceManager(flow)
+		return
+	}
+
 	var updatedFlows []rsrcMgr.FlowInfo
 	var flowID uint32
 	var onuID, uniID int32
@@ -1570,6 +1624,61 @@
 	}
 }
 
+//clearMulticastFlowFromResourceManager  removes a multicast flow from the KV store and
+// clears resources reserved for this multicast flow
+func (f *OpenOltFlowMgr) clearMulticastFlowFromResourceManager(flow *ofp.OfpFlowStats) {
+	classifierInfo := make(map[string]interface{})
+	formulateClassifierInfoFromFlow(classifierInfo, flow)
+	inPort, err := f.getInPortOfMulticastFlow(classifierInfo)
+
+	if err != nil {
+		log.Warnw("No inPort found. Cannot release resources of the multicast flow.", log.Fields{"flowId:": flow.Id})
+		return
+	}
+
+	networkInterfaceID := IntfIDFromNniPortNum(inPort)
+	var onuID = int32(NoneOnuID)
+	var uniID = int32(NoneUniID)
+	var flowID uint32
+	var updatedFlows []rsrcMgr.FlowInfo
+
+	flowIds := f.resourceMgr.GetCurrentFlowIDsForOnu(networkInterfaceID, onuID, uniID)
+
+	for _, flowID = range flowIds {
+		flowInfo := f.resourceMgr.GetFlowIDInfo(networkInterfaceID, onuID, uniID, flowID)
+		if flowInfo == nil {
+			log.Debugw("No multicast FlowInfo found in the KV store",
+				log.Fields{"Intf": networkInterfaceID, "onuID": onuID, "uniID": uniID, "flowID": flowID})
+			continue
+		}
+		updatedFlows = nil
+		for _, flow := range *flowInfo {
+			updatedFlows = append(updatedFlows, flow)
+		}
+		for i, storedFlow := range updatedFlows {
+			if flow.Id == storedFlow.LogicalFlowID {
+				removeFlowMessage := openoltpb2.Flow{FlowId: storedFlow.Flow.FlowId, FlowType: storedFlow.Flow.FlowType}
+				log.Debugw("Multicast flow to be deleted", log.Fields{"flow": storedFlow})
+				//remove from device
+				if ok := f.removeFlowFromDevice(&removeFlowMessage); !ok {
+					log.Errorw("Failed to remove multicast flow from device", log.Fields{"flowId": flow.Id})
+					return
+				}
+				log.Debugw("Multicast flow removed from device successfully", log.Fields{"flowId": flow.Id})
+				//Remove the Flow from FlowInfo
+				updatedFlows = append(updatedFlows[:i], updatedFlows[i+1:]...)
+				if err := f.updateFlowInfoToKVStore(int32(networkInterfaceID), NoneOnuID, NoneUniID, flowID, &updatedFlows); err != nil {
+					log.Error("Failed to delete multicast flow from the KV store", log.Fields{"flow": storedFlow, "err": err})
+					return
+				}
+				//release flow id
+				log.Debugw("Releasing multicast flow id", log.Fields{"flowId": flowID, "interfaceID": networkInterfaceID})
+				f.resourceMgr.FreeFlowID(uint32(networkInterfaceID), NoneOnuID, NoneUniID, flowID)
+			}
+		}
+	}
+}
+
 //RemoveFlow removes the flow from the device
 func (f *OpenOltFlowMgr) RemoveFlow(flow *ofp.OfpFlowStats) {
 	log.Debugw("Removing Flow", log.Fields{"flow": flow})
@@ -1587,12 +1696,14 @@
 			}
 		}
 	}
-	if IsUpstream(actionInfo[Output].(uint32)) {
+
+	if flows.HasGroup(flow) {
+		direction = Multicast
+	} else if IsUpstream(actionInfo[Output].(uint32)) {
 		direction = Upstream
 	} else {
 		direction = Downstream
 	}
-
 	f.clearFlowFromResourceManager(flow, direction) //TODO: Take care of the limitations
 
 	return
@@ -1650,6 +1761,12 @@
 		return
 	}
 
+	if flows.HasGroup(flow) {
+		// handle multicast flow
+		f.handleFlowWithGroup(actionInfo, classifierInfo, flow)
+		return
+	}
+
 	/* Controller bound trap flows */
 	err = formulateControllerBoundTrapFlowInfo(actionInfo, classifierInfo, flow)
 	if err != nil {
@@ -1723,6 +1840,283 @@
 	}
 }
 
+// handleFlowWithGroup adds multicast flow to the device.
+func (f *OpenOltFlowMgr) handleFlowWithGroup(actionInfo, classifierInfo map[string]interface{}, flow *ofp.OfpFlowStats) {
+	classifierInfo[PacketTagType] = DoubleTag
+	log.Debugw("add-multicast-flow", log.Fields{"classifierInfo": classifierInfo, "actionInfo": actionInfo})
+
+	inPort, err := f.getInPortOfMulticastFlow(classifierInfo)
+	if err != nil {
+		log.Warnw("No inPort found. Ignoring multicast flow.", log.Fields{"flowId:": flow.Id})
+		return
+	}
+	//replace ipDst with ethDst
+	if ipv4Dst, ok := classifierInfo[Ipv4Dst]; ok &&
+		flows.IsMulticastIp(ipv4Dst.(uint32)) {
+		// replace ipv4_dst classifier with eth_dst
+		multicastMac := flows.ConvertToMulticastMacBytes(ipv4Dst.(uint32))
+		delete(classifierInfo, Ipv4Dst)
+		delete(classifierInfo, EthType)
+		classifierInfo[EthDst] = multicastMac
+		log.Debugw("multicast-ip-to-mac-conversion-success", log.Fields{"ip:": ipv4Dst.(uint32), "mac:": multicastMac})
+	}
+
+	var onuID = NoneOnuID
+	var uniID = NoneUniID
+	var gemPortID = NoneGemPortID
+
+	networkInterfaceID := IntfIDFromNniPortNum(inPort)
+
+	var flowStoreCookie = getFlowStoreCookie(classifierInfo, uint32(0))
+	if present := f.resourceMgr.IsFlowCookieOnKVStore(uint32(networkInterfaceID), int32(onuID), int32(uniID), flowStoreCookie); present {
+		log.Debugw("multicast-flow-exists--not-re-adding", log.Fields{"classifierInfo": classifierInfo})
+		return
+	}
+	flowID, err := f.resourceMgr.GetFlowID(uint32(networkInterfaceID), int32(onuID), int32(uniID), uint32(gemPortID), flowStoreCookie, "", 0, 0)
+	if err != nil {
+		log.Errorw("Flow id unavailable for multicast flow", log.Fields{"error": err})
+		return
+	}
+	var classifierProto *openoltpb2.Classifier
+	if classifierProto = makeOpenOltClassifierField(classifierInfo); classifierProto == nil {
+		log.Error("Error in making classifier protobuf for multicast flow")
+		return
+	}
+	groupID := actionInfo[GroupID].(uint32)
+	multicastFlow := openoltpb2.Flow{
+		FlowId:        flowID,
+		FlowType:      Multicast,
+		NetworkIntfId: int32(networkInterfaceID),
+		GroupId:       groupID,
+		Classifier:    classifierProto,
+		Priority:      int32(flow.Priority),
+		Cookie:        flow.Cookie}
+
+	if ok := f.addFlowToDevice(flow, &multicastFlow); ok {
+		log.Debug("multicast flow added to device successfully")
+		//get cached group
+		group, _, err := f.GetFlowGroupFromKVStore(groupID, true)
+		if err == nil {
+			//calling groupAdd to set group members after multicast flow creation
+			if f.ModifyGroup(group) {
+				//cached group can be removed now
+				f.resourceMgr.RemoveFlowGroupFromKVStore(groupID, true)
+			}
+		}
+
+		flowsToKVStore := f.getUpdatedFlowInfo(&multicastFlow, flowStoreCookie, MulticastFlow, flowID, flow.Id)
+		if err := f.updateFlowInfoToKVStore(int32(networkInterfaceID),
+			int32(onuID),
+			int32(uniID),
+			flowID, flowsToKVStore); err != nil {
+			log.Errorw("Error uploading multicast flow into KV store", log.Fields{"flow": multicastFlow, "error": err})
+		}
+	}
+	return
+}
+
+//getInPortOfMulticastFlow return inPort criterion if exists; returns NNI interface of the device otherwise
+func (f *OpenOltFlowMgr) getInPortOfMulticastFlow(classifierInfo map[string]interface{}) (uint32, error) {
+	if _, ok := classifierInfo[InPort]; ok {
+		return classifierInfo[InPort].(uint32), nil
+	}
+	// find first NNI port of the device
+	nniPorts, e := f.resourceMgr.GetNNIFromKVStore()
+	if e == nil && len(nniPorts) > 0 {
+		return nniPorts[0], nil
+	}
+	return 0, errors.New("cannot find NNI port of device")
+}
+
+// AddGroup add or update the group
+func (f *OpenOltFlowMgr) AddGroup(group *ofp.OfpGroupEntry) {
+	log.Infow("add-group", log.Fields{"group": group})
+	if group == nil {
+		log.Warn("skipping nil group")
+		return
+	}
+
+	groupToOlt := openoltpb2.Group{
+		GroupId: group.Desc.GroupId,
+		Command: openoltpb2.Group_SET_MEMBERS,
+		Action:  f.buildGroupAction(),
+	}
+
+	log.Debugw("Sending group to device", log.Fields{"groupToOlt": groupToOlt})
+	_, err := f.deviceHandler.Client.PerformGroupOperation(context.Background(), &groupToOlt)
+	if err != nil {
+		log.Errorw("add-group operation failed", log.Fields{"err": err, "groupToOlt": groupToOlt})
+		return
+	}
+	// group members not created yet. So let's store the group
+	if err := f.resourceMgr.AddFlowGroupToKVStore(group, true); err != nil {
+		log.Errorw("Group cannot be stored in KV store", log.Fields{"groupId": group.Desc.GroupId, "err": err})
+	} else {
+		log.Debugw("add-group operation performed on the device successfully ", log.Fields{"groupToOlt": groupToOlt})
+	}
+}
+
+//buildGroupAction creates and returns a group action
+func (f *OpenOltFlowMgr) buildGroupAction() *openoltpb2.Action {
+	var actionCmd openoltpb2.ActionCmd
+	var action openoltpb2.Action
+	action.Cmd = &actionCmd
+	//pop outer vlan
+	action.Cmd.RemoveOuterTag = true
+	return &action
+}
+
+// ModifyGroup updates the group
+func (f *OpenOltFlowMgr) ModifyGroup(group *ofp.OfpGroupEntry) bool {
+	log.Infow("modify-group", log.Fields{"group": group})
+	if group == nil || group.Desc == nil {
+		log.Warn("cannot modify group; group is nil")
+		return false
+	}
+
+	new := f.buildGroup(group.Desc.GroupId, group.Desc.Buckets)
+	//get existing members of the group
+	val, groupExists, err := f.GetFlowGroupFromKVStore(group.Desc.GroupId, false)
+
+	if err != nil {
+		log.Errorw("Failed to retrieve the group from the store. Cannot modify group.",
+			log.Fields{"groupId": group.Desc.GroupId, "err": err})
+		return false
+	}
+
+	var current *openoltpb2.Group
+	if groupExists {
+		// group already exists
+		current = f.buildGroup(group.Desc.GroupId, val.Desc.GetBuckets())
+		log.Debugw("modify-group: group exists.", log.Fields{"current": val, "new": group})
+	} else {
+		current = f.buildGroup(group.Desc.GroupId, nil)
+	}
+
+	log.Debugw("modify-group: comparing current and new.", log.Fields{"current": current, "new": new})
+	// check if the buckets are identical
+	bucketsIdentical := f.bucketsIdentical(current, new)
+
+	isSuccess := true
+	if !bucketsIdentical {
+		groupToOlt := openoltpb2.Group{
+			GroupId: group.Desc.GroupId,
+			Command: openoltpb2.Group_SET_MEMBERS,
+			Members: new.Members,
+			Action:  f.buildGroupAction(),
+		}
+
+		if err := f.callGroupAdd(&groupToOlt); err != nil {
+			log.Warnw("One of the group add/remove operations has failed. Cannot save group modifications",
+				log.Fields{"group": group})
+			isSuccess = false
+		}
+	}
+
+	if isSuccess {
+		if err := f.resourceMgr.AddFlowGroupToKVStore(group, false); err != nil {
+			log.Errorw("Failed to save the group into kv store", log.Fields{"groupId": group.Desc.GroupId})
+		}
+		log.Debugw("modify-group was success. Storing the group", log.Fields{"group": group, "existingGroup": current})
+	}
+	return isSuccess
+}
+
+//bucketsIdentical returns true if groups are identical; false otherwise
+func (f *OpenOltFlowMgr) bucketsIdentical(current *openoltpb2.Group, new *openoltpb2.Group) bool {
+	if current.GroupId == new.GroupId &&
+		len(new.Members) == len(current.Members) {
+		diff := f.findDiff(current, new)
+		if diff == nil || len(diff) < 1 {
+			log.Infow("modify-group: current and new buckets are the same. Won't send SET_MEMBERS again.",
+				log.Fields{"groupId:": current.GroupId})
+			return true
+		}
+	}
+	return false
+}
+
+//findDiff compares group members and finds members which only exists in groups2
+func (f *OpenOltFlowMgr) findDiff(group1 *openoltpb2.Group, group2 *openoltpb2.Group) []*openoltpb2.GroupMember {
+	var members []*openoltpb2.GroupMember
+	for _, bucket := range group2.Members {
+		if !f.contains(group1.Members, bucket) {
+			// bucket does not exist and must be added
+			members = append(members, bucket)
+		}
+	}
+	return members
+}
+
+//contains returns true if the members list contains the given member; false otherwise
+func (f *OpenOltFlowMgr) contains(members []*openoltpb2.GroupMember, member *openoltpb2.GroupMember) bool {
+	for _, groupMember := range members {
+		if groupMember.InterfaceId == member.InterfaceId {
+			return true
+		}
+	}
+	return false
+}
+
+//callGroupAdd call GroupAdd operation of openolt proto
+func (f *OpenOltFlowMgr) callGroupAdd(group *openoltpb2.Group) error {
+	log.Debugw("Sending group to device", log.Fields{"groupToOlt": group, "command": group.Command})
+	_, err := f.deviceHandler.Client.PerformGroupOperation(context.Background(), group)
+	if err != nil {
+		log.Errorw("group operation failed", log.Fields{"err": err, "groupToOlt": group})
+	}
+	return err
+}
+
+//buildGroup build openoltpb2.Group from given group id and bucket list
+func (f *OpenOltFlowMgr) buildGroup(groupID uint32, buckets []*ofp.OfpBucket) *openoltpb2.Group {
+	group := openoltpb2.Group{
+		GroupId: groupID}
+	// create members of the group
+	if buckets != nil {
+		for _, ofBucket := range buckets {
+			member := f.buildMember(ofBucket)
+			if member != nil && !f.contains(group.Members, member) {
+				group.Members = append(group.Members, member)
+			}
+		}
+	}
+	return &group
+}
+
+//buildMember builds openoltpb2.GroupMember from an OpenFlow bucket
+func (f *OpenOltFlowMgr) buildMember(ofBucket *ofp.OfpBucket) *openoltpb2.GroupMember {
+	var outPort uint32
+	outPortFound := false
+	for _, ofAction := range ofBucket.Actions {
+		if ofAction.Type == ofp.OfpActionType_OFPAT_OUTPUT {
+			outPort = ofAction.GetOutput().Port
+			outPortFound = true
+		}
+	}
+
+	if !outPortFound {
+		log.Debugw("bucket skipped since no out port found in it",
+			log.Fields{"ofBucket": ofBucket})
+		return nil
+	}
+	interfaceID := IntfIDFromUniPortNum(outPort)
+	log.Debugw("got associated interface id of the port", log.Fields{"portNumber:": outPort, "interfaceId:": interfaceID})
+	if groupInfo, ok := f.interfaceToMcastQueueMap[interfaceID]; ok {
+		member := openoltpb2.GroupMember{
+			InterfaceId:   interfaceID,
+			InterfaceType: openoltpb2.GroupMember_PON,
+			GemPortId:     groupInfo.gemPortID,
+			Priority:      groupInfo.servicePriority,
+		}
+		//add member to the group
+		return &member
+	}
+	log.Warnf("bucket skipped since interface-2-gem mapping cannot be found",
+		log.Fields{"ofBucket": ofBucket})
+	return nil
+}
+
 //sendTPDownloadMsgToChild send payload
 func (f *OpenOltFlowMgr) sendTPDownloadMsgToChild(intfID uint32, onuID uint32, uniID uint32, uni string, TpID uint32) error {
 
@@ -2228,6 +2622,9 @@
 		if field.Type == flows.ETH_TYPE {
 			classifierInfo[EthType] = field.GetEthType()
 			log.Debug("field-type-eth-type", log.Fields{"classifierInfo[ETH_TYPE]": classifierInfo[EthType].(uint32)})
+		} else if field.Type == flows.ETH_DST {
+			classifierInfo[EthDst] = field.GetEthDst()
+			log.Debug("field-type-eth-type", log.Fields{"classifierInfo[ETH_DST]": classifierInfo[EthDst].([]uint8)})
 		} else if field.Type == flows.IP_PROTO {
 			classifierInfo[IPProto] = field.GetIpProto()
 			log.Debug("field-type-ip-proto", log.Fields{"classifierInfo[IP_PROTO]": classifierInfo[IPProto].(uint32)})
@@ -2297,20 +2694,11 @@
 						return errors.New("invalid openflow class")
 					}
 					/*log.Debugw("action-type-set-field",log.Fields{"field": field, "in_port": classifierInfo[IN_PORT].(uint32)})*/
-					if ofbField := field.GetOfbField(); ofbField != nil {
-						if fieldtype := ofbField.GetType(); fieldtype == ofp.OxmOfbFieldTypes_OFPXMT_OFB_VLAN_VID {
-							if vlan := ofbField.GetVlanVid(); vlan != 0 {
-								actionInfo[VlanVid] = vlan & 0xfff
-								log.Debugw("action-set-vlan-vid", log.Fields{"actionInfo[VLAN_VID]": actionInfo[VlanVid].(uint32)})
-							} else {
-								log.Error("No Invalid vlan id in set vlan-vid action")
-							}
-						} else {
-							log.Errorw("unsupported-action-set-field-type", log.Fields{"type": fieldtype})
-						}
-					}
+					formulateSetFieldActionInfoFromFlow(field, actionInfo)
 				}
 			}
+		} else if action.Type == flows.GROUP {
+			formulateGroupActionInfoFromFlow(action, actionInfo)
 		} else {
 			log.Errorw("Un supported action type", log.Fields{"type": action.Type})
 			return errors.New("un supported action type")
@@ -2319,6 +2707,30 @@
 	return nil
 }
 
+func formulateSetFieldActionInfoFromFlow(field *ofp.OfpOxmField, actionInfo map[string]interface{}) {
+	if ofbField := field.GetOfbField(); ofbField != nil {
+		if fieldtype := ofbField.GetType(); fieldtype == ofp.OxmOfbFieldTypes_OFPXMT_OFB_VLAN_VID {
+			if vlan := ofbField.GetVlanVid(); vlan != 0 {
+				actionInfo[VlanVid] = vlan & 0xfff
+				log.Debugw("action-set-vlan-vid", log.Fields{"actionInfo[VLAN_VID]": actionInfo[VlanVid].(uint32)})
+			} else {
+				log.Error("No Invalid vlan id in set vlan-vid action")
+			}
+		} else {
+			log.Errorw("unsupported-action-set-field-type", log.Fields{"type": fieldtype})
+		}
+	}
+}
+
+func formulateGroupActionInfoFromFlow(action *ofp.OfpAction, actionInfo map[string]interface{}) {
+	if action.GetGroup() == nil {
+		log.Warn("No group entry found in the group action")
+	} else {
+		actionInfo[GroupID] = action.GetGroup().GroupId
+		log.Debugw("action-group-id", log.Fields{"actionInfo[GroupID]": actionInfo[GroupID].(uint32)})
+	}
+}
+
 func formulateControllerBoundTrapFlowInfo(actionInfo, classifierInfo map[string]interface{}, flow *ofp.OfpFlowStats) error {
 	if isControllerFlow := IsControllerBoundFlow(actionInfo[Output].(uint32)); isControllerFlow {
 		log.Debug("Controller bound trap flows, getting inport from tunnelid")
@@ -2456,3 +2868,54 @@
 	}
 	return
 }
+
+//loadInterfaceToMulticastQueueMap reads multicast queues per interface from the KV store
+//and put them into interfaceToMcastQueueMap.
+func (f *OpenOltFlowMgr) loadInterfaceToMulticastQueueMap() {
+	storedMulticastQueueMap, err := f.resourceMgr.GetMcastQueuePerInterfaceMap()
+	if err != nil {
+		log.Error("Failed to get pon interface to multicast queue map")
+		return
+	}
+	for intf, queueInfo := range storedMulticastQueueMap {
+		q := queueInfoBrief{
+			gemPortID:       queueInfo[0],
+			servicePriority: queueInfo[1],
+		}
+		f.interfaceToMcastQueueMap[intf] = &q
+	}
+}
+
+//GetFlowGroupFromKVStore fetches and returns flow group from the KV store. Returns (nil, false, error) if any problem occurs during
+//fetching the data. Returns (group, true, nil) if the group is fetched and returned successfully.
+//Returns (nil, false, nil) if the group does not exists in the KV store.
+func (f *OpenOltFlowMgr) GetFlowGroupFromKVStore(groupID uint32, cached bool) (*ofp.OfpGroupEntry, bool, error) {
+	exists, groupInfo, err := f.resourceMgr.GetFlowGroupFromKVStore(groupID, cached)
+	if err != nil {
+		log.Errorw("Failed to get the flow group from KV store", log.Fields{"groupId": groupID, "err": err})
+		return nil, false, errors.New("failed to retrieve the flow group")
+	}
+	if exists {
+		return newGroup(groupInfo.GroupID, groupInfo.OutPorts), exists, nil
+	}
+	return nil, exists, nil
+}
+
+func newGroup(groupID uint32, outPorts []uint32) *ofp.OfpGroupEntry {
+	groupDesc := ofp.OfpGroupDesc{
+		Type:    ofp.OfpGroupType_OFPGT_ALL,
+		GroupId: groupID,
+	}
+	groupEntry := ofp.OfpGroupEntry{
+		Desc: &groupDesc,
+	}
+	var acts []*ofp.OfpAction
+	for i := 0; i < len(outPorts); i++ {
+		acts = append(acts, flows.Output(outPorts[i]))
+	}
+	bucket := ofp.OfpBucket{
+		Actions: acts,
+	}
+	groupDesc.Buckets = []*ofp.OfpBucket{&bucket}
+	return &groupEntry
+}