diff --git a/.gen/proto/sharddistributor/v1/executor.pb.go b/.gen/proto/sharddistributor/v1/executor.pb.go
index 44bd7965f09..1a369186b18 100644
--- a/.gen/proto/sharddistributor/v1/executor.pb.go
+++ b/.gen/proto/sharddistributor/v1/executor.pb.go
@@ -156,6 +156,7 @@ type HeartbeatRequest struct {
 	ExecutorId           string                        `protobuf:"bytes,2,opt,name=executor_id,json=executorId,proto3" json:"executor_id,omitempty"`
 	Status               ExecutorStatus                `protobuf:"varint,3,opt,name=status,proto3,enum=uber.cadence.sharddistributor.v1.ExecutorStatus" json:"status,omitempty"`
 	ShardStatusReports   map[string]*ShardStatusReport `protobuf:"bytes,4,rep,name=shard_status_reports,json=shardStatusReports,proto3" json:"shard_status_reports,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
+	Metadata             map[string]string             `protobuf:"bytes,5,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
 	XXX_NoUnkeyedLiteral struct{}                      `json:"-"`
 	XXX_unrecognized     []byte                        `json:"-"`
 	XXX_sizecache        int32                         `json:"-"`
@@ -222,6 +223,13 @@ func (m *HeartbeatRequest) GetShardStatusReports() map[string]*ShardStatusReport
 	return nil
 }
 
+func (m *HeartbeatRequest) GetMetadata() map[string]string {
+	if m != nil {
+		return m.Metadata
+	}
+	return nil
+}
+
 type ShardStatusReport struct {
 	Status               ShardStatus `protobuf:"varint,1,opt,name=status,proto3,enum=uber.cadence.sharddistributor.v1.ShardStatus" json:"status,omitempty"`
 	ShardLoad            float64     `protobuf:"fixed64,2,opt,name=shard_load,json=shardLoad,proto3" json:"shard_load,omitempty"`
@@ -385,6 +393,7 @@ func init() {
 	proto.RegisterEnum("uber.cadence.sharddistributor.v1.AssignmentStatus", AssignmentStatus_name, AssignmentStatus_value)
 	proto.RegisterEnum("uber.cadence.sharddistributor.v1.MigrationMode", MigrationMode_name, MigrationMode_value)
 	proto.RegisterType((*HeartbeatRequest)(nil), "uber.cadence.sharddistributor.v1.HeartbeatRequest")
+	proto.RegisterMapType((map[string]string)(nil), "uber.cadence.sharddistributor.v1.HeartbeatRequest.MetadataEntry")
 	proto.RegisterMapType((map[string]*ShardStatusReport)(nil), "uber.cadence.sharddistributor.v1.HeartbeatRequest.ShardStatusReportsEntry")
 	proto.RegisterType((*ShardStatusReport)(nil), "uber.cadence.sharddistributor.v1.ShardStatusReport")
 	proto.RegisterType((*HeartbeatResponse)(nil), "uber.cadence.sharddistributor.v1.HeartbeatResponse")
@@ -397,52 +406,55 @@ func init() {
 }
 
 var fileDescriptor_5aab034437d08cca = []byte{
-	// 713 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x55, 0xcd, 0x6e, 0xda, 0x4c,
-	0x14, 0xfd, 0x06, 0xf2, 0x45, 0xe2, 0xa2, 0x50, 0x33, 0xca, 0x0f, 0x25, 0x3f, 0x45, 0x51, 0xd5,
-	0x46, 0x54, 0x35, 0x85, 0x6c, 0xaa, 0x76, 0xe5, 0xc4, 0x16, 0x4c, 0x04, 0x76, 0x34, 0x36, 0xa4,
-	0xad, 0x54, 0x59, 0x06, 0x8f, 0x88, 0xd5, 0x60, 0x53, 0xdb, 0xa0, 0xa6, 0xea, 0xb2, 0x6f, 0x50,
-	0xf5, 0x9d, 0xb2, 0xec, 0xbe, 0x9b, 0x2a, 0x4f, 0x52, 0xd9, 0x86, 0x80, 0x0d, 0x15, 0x4d, 0x77,
-	0xf6, 0x3d, 0xf7, 0x9c, 0x3b, 0x3e, 0x67, 0xc6, 0x03, 0x95, 0x51, 0x97, 0xb9, 0x95, 0x9e, 0x61,
-	0x32, 0xbb, 0xc7, 0x2a, 0xde, 0xa5, 0xe1, 0x9a, 0xa6, 0xe5, 0xf9, 0xae, 0xd5, 0x1d, 0xf9, 0x8e,
-	0x5b, 0x19, 0x57, 0x2b, 0xec, 0x13, 0xeb, 0x05, 0xcf, 0xfc, 0xd0, 0x75, 0x7c, 0x07, 0x97, 0x02,
-	0x02, 0x3f, 0x21, 0xf0, 0x49, 0x02, 0x3f, 0xae, 0x1e, 0x7e, 0x4f, 0x03, 0xd7, 0x60, 0x86, 0xeb,
-	0x77, 0x99, 0xe1, 0x53, 0xf6, 0x71, 0xc4, 0x3c, 0x1f, 0xef, 0x41, 0xc6, 0x36, 0x06, 0xcc, 0x1b,
-	0x1a, 0x3d, 0x56, 0x40, 0x25, 0x74, 0x94, 0xa1, 0xb3, 0x02, 0x7e, 0x04, 0xd9, 0xe9, 0x18, 0xdd,
-	0x32, 0x0b, 0xa9, 0x10, 0x87, 0x69, 0x89, 0x98, 0xb8, 0x01, 0xeb, 0x9e, 0x6f, 0xf8, 0x23, 0xaf,
-	0x90, 0x2e, 0xa1, 0xa3, 0x5c, 0xed, 0x05, 0xbf, 0x6a, 0x19, 0xbc, 0x34, 0x61, 0xab, 0x21, 0x8f,
-	0x4e, 0xf8, 0xf8, 0x0b, 0x6c, 0x86, 0xdd, 0x7a, 0xf4, 0xae, 0xbb, 0x6c, 0xe8, 0xb8, 0xbe, 0x57,
-	0x58, 0x2b, 0xa5, 0x8f, 0xb2, 0xb5, 0xb3, 0xd5, 0xba, 0xc9, 0x4f, 0xe3, 0xd5, 0xa0, 0x69, 0x32,
-	0x25, 0x12, 0x93, 0x6c, 0xdf, 0xbd, 0xa6, 0xd8, 0x5b, 0x00, 0x8a, 0x9f, 0x61, 0xe7, 0x0f, 0xed,
-	0x98, 0x83, 0xf4, 0x07, 0x76, 0x3d, 0xf1, 0x26, 0x78, 0xc4, 0x04, 0xfe, 0x1f, 0x1b, 0x57, 0x23,
-	0x16, 0xfa, 0x91, 0xad, 0x1d, 0xaf, 0x5e, 0xdb, 0x82, 0x36, 0x8d, 0x14, 0x5e, 0xa5, 0x5e, 0xa2,
-	0xc3, 0x6b, 0xc8, 0x2f, 0xe0, 0x58, 0xba, 0x33, 0x16, 0x85, 0xc6, 0x3e, 0xbf, 0xdf, 0x90, 0xa9,
-	0xab, 0xfb, 0x00, 0x91, 0xab, 0x57, 0x8e, 0x11, 0xe5, 0x87, 0x68, 0x26, 0xac, 0x34, 0x1d, 0xc3,
-	0x3c, 0xfc, 0x99, 0x82, 0xfc, 0x9c, 0x6f, 0xde, 0xd0, 0xb1, 0x3d, 0x86, 0xc7, 0x90, 0x8f, 0x48,
-	0x86, 0xe7, 0x59, 0x7d, 0x7b, 0xc0, 0x6c, 0x3f, 0x58, 0x46, 0x90, 0x03, 0xb9, 0x57, 0x0e, 0x91,
-	0x5e, 0xb4, 0x30, 0x61, 0xa6, 0x15, 0xc5, 0xc0, 0x79, 0x89, 0x32, 0xee, 0x40, 0x6e, 0x60, 0xf5,
-	0x5d, 0xc3, 0xb7, 0x1c, 0x5b, 0x1f, 0x38, 0x66, 0x64, 0x70, 0xae, 0x56, 0x59, 0x3d, 0xb4, 0x35,
-	0xe5, 0xb5, 0x1c, 0x93, 0xd1, 0x8d, 0xc1, 0xfc, 0x6b, 0x71, 0x0c, 0x5b, 0x4b, 0x97, 0xb0, 0x24,
-	0xda, 0x7a, 0x3c, 0xda, 0xea, 0x5f, 0xba, 0x3e, 0x53, 0x9e, 0x0f, 0xf6, 0x3d, 0x3c, 0x48, 0xa0,
-	0xf8, 0x2c, 0x11, 0x6b, 0x6d, 0xf5, 0x80, 0x19, 0x3b, 0x9e, 0x6d, 0xf9, 0x2b, 0x82, 0x5c, 0xfc,
-	0x30, 0xe1, 0x5d, 0xd8, 0x91, 0xde, 0x48, 0xa7, 0x6d, 0x4d, 0xa1, 0xba, 0xaa, 0x09, 0x5a, 0x5b,
-	0xd5, 0x89, 0xdc, 0x11, 0x9a, 0x44, 0xe4, 0xfe, 0xc3, 0x45, 0xd8, 0x4e, 0x82, 0xc2, 0xa9, 0x46,
-	0x3a, 0x12, 0x87, 0xf0, 0x1e, 0x14, 0x92, 0x98, 0x48, 0x05, 0x22, 0x13, 0xb9, 0xce, 0xa5, 0x96,
-	0xc9, 0x86, 0xa8, 0x24, 0x72, 0xe9, 0x72, 0x07, 0xb2, 0x73, 0x3b, 0x0f, 0x17, 0x60, 0x53, 0x6d,
-	0x08, 0x54, 0x5c, 0x9c, 0xbf, 0x0d, 0x38, 0x86, 0x50, 0x49, 0x10, 0xdf, 0x72, 0x08, 0x6f, 0x41,
-	0x3e, 0x56, 0x17, 0x15, 0x59, 0xe2, 0x52, 0x65, 0x19, 0xb8, 0xe4, 0xa7, 0xe3, 0x7d, 0x78, 0x28,
-	0xa8, 0x2a, 0xa9, 0xcb, 0x2d, 0x49, 0xd6, 0x16, 0x27, 0xec, 0xc2, 0xce, 0x22, 0x3c, 0x19, 0x53,
-	0xbe, 0x41, 0xb0, 0x11, 0xdb, 0x26, 0x81, 0x21, 0x2d, 0x52, 0xa7, 0x82, 0x46, 0x14, 0x59, 0x6f,
-	0x29, 0xa2, 0x34, 0x27, 0xf5, 0x18, 0x4a, 0x09, 0xac, 0xa9, 0x9c, 0x0a, 0x4d, 0xfd, 0x5c, 0x50,
-	0x55, 0xad, 0x41, 0x95, 0x76, 0xbd, 0xc1, 0x21, 0xfc, 0x0c, 0x9e, 0xae, 0xea, 0xd2, 0xd5, 0x86,
-	0x20, 0x2a, 0x17, 0x5c, 0x0a, 0x97, 0xe1, 0x49, 0xa2, 0x59, 0x24, 0xaa, 0x46, 0xc9, 0x49, 0x5b,
-	0x93, 0xc4, 0x98, 0x70, 0x3a, 0xc8, 0x23, 0xd1, 0xab, 0xc8, 0x27, 0x8a, 0x40, 0x45, 0x49, 0xe4,
-	0xd6, 0x6a, 0xdf, 0x10, 0xec, 0x86, 0x9e, 0x8b, 0xb3, 0xad, 0x32, 0xdd, 0x09, 0xc2, 0x39, 0xc1,
-	0x3e, 0x64, 0xee, 0x4e, 0x21, 0xae, 0xdd, 0xff, 0xd7, 0x59, 0x3c, 0xfe, 0x87, 0x63, 0x7e, 0x72,
-	0x71, 0x73, 0x7b, 0x80, 0x7e, 0xdc, 0x1e, 0xa0, 0x5f, 0xb7, 0x07, 0xe8, 0x1d, 0xe9, 0x5b, 0xfe,
-	0xe5, 0xa8, 0xcb, 0xf7, 0x9c, 0x41, 0xfc, 0x2e, 0xe3, 0xfb, 0xcc, 0xae, 0x84, 0x77, 0xd6, 0xb2,
-	0x6b, 0xed, 0x75, 0xb2, 0x36, 0xae, 0x76, 0xd7, 0xc3, 0xee, 0xe3, 0xdf, 0x01, 0x00, 0x00, 0xff,
-	0xff, 0x83, 0x41, 0x28, 0x37, 0x14, 0x07, 0x00, 0x00,
+	// 754 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x55, 0xcd, 0x6e, 0xda, 0x4a,
+	0x14, 0xbe, 0x86, 0x24, 0xba, 0x1c, 0x14, 0xae, 0x19, 0xe5, 0x87, 0x4b, 0x7e, 0x2e, 0x8a, 0xae,
+	0xda, 0x88, 0xaa, 0xa6, 0x90, 0x4d, 0xd5, 0x6c, 0xea, 0xc4, 0x16, 0x38, 0x02, 0x3b, 0x1a, 0x1b,
+	0xd2, 0x56, 0xad, 0xac, 0x01, 0x8f, 0x08, 0x6a, 0xb0, 0xa9, 0x3d, 0xa0, 0xa6, 0xea, 0xb2, 0x6f,
+	0xd0, 0x17, 0xe9, 0x63, 0x64, 0xd9, 0x7d, 0x37, 0x55, 0x9e, 0xa4, 0xb2, 0x0d, 0x01, 0x1b, 0x2a,
+	0x9a, 0xec, 0xec, 0x73, 0xce, 0xf7, 0x7d, 0xc7, 0xe7, 0x3b, 0xe3, 0x81, 0xd2, 0xb0, 0x4d, 0xdd,
+	0x52, 0x87, 0x58, 0xd4, 0xee, 0xd0, 0x92, 0x77, 0x49, 0x5c, 0xcb, 0xea, 0x79, 0xcc, 0xed, 0xb5,
+	0x87, 0xcc, 0x71, 0x4b, 0xa3, 0x72, 0x89, 0x7e, 0xa4, 0x1d, 0xff, 0x59, 0x18, 0xb8, 0x0e, 0x73,
+	0x50, 0xc1, 0x07, 0x08, 0x63, 0x80, 0x10, 0x07, 0x08, 0xa3, 0xf2, 0xc1, 0xb7, 0x15, 0xe0, 0x6b,
+	0x94, 0xb8, 0xac, 0x4d, 0x09, 0xc3, 0xf4, 0xc3, 0x90, 0x7a, 0x0c, 0xed, 0x42, 0xca, 0x26, 0x7d,
+	0xea, 0x0d, 0x48, 0x87, 0xe6, 0xb8, 0x02, 0x77, 0x98, 0xc2, 0xd3, 0x00, 0xfa, 0x0f, 0xd2, 0x13,
+	0x19, 0xb3, 0x67, 0xe5, 0x12, 0x41, 0x1e, 0x26, 0x21, 0xc5, 0x42, 0x35, 0x58, 0xf3, 0x18, 0x61,
+	0x43, 0x2f, 0x97, 0x2c, 0x70, 0x87, 0x99, 0xca, 0x33, 0x61, 0x59, 0x1b, 0x82, 0x3c, 0x46, 0xeb,
+	0x01, 0x0e, 0x8f, 0xf1, 0xe8, 0x33, 0x6c, 0x04, 0xd5, 0x66, 0xf8, 0x6e, 0xba, 0x74, 0xe0, 0xb8,
+	0xcc, 0xcb, 0xad, 0x14, 0x92, 0x87, 0xe9, 0xca, 0xd9, 0x72, 0xde, 0xf8, 0xa7, 0x09, 0xba, 0x5f,
+	0x34, 0x56, 0x09, 0xc9, 0x64, 0x9b, 0xb9, 0xd7, 0x18, 0x79, 0x73, 0x09, 0xf4, 0x16, 0xfe, 0xee,
+	0x53, 0x46, 0x2c, 0xc2, 0x48, 0x6e, 0x35, 0x50, 0x7c, 0xf9, 0x00, 0xc5, 0xc6, 0x98, 0x22, 0xd4,
+	0xb9, 0x63, 0xcc, 0x7f, 0x82, 0xed, 0xdf, 0x34, 0x83, 0x78, 0x48, 0xbe, 0xa7, 0xd7, 0xe3, 0xc9,
+	0xfb, 0x8f, 0x48, 0x81, 0xd5, 0x11, 0xb9, 0x1a, 0xd2, 0x60, 0xda, 0xe9, 0xca, 0xd1, 0xf2, 0x3e,
+	0xe6, 0xb8, 0x71, 0xc8, 0xf0, 0x22, 0xf1, 0x9c, 0xcb, 0x1f, 0xc3, 0x7a, 0xa4, 0xad, 0x05, 0x8a,
+	0x1b, 0xb3, 0x8a, 0xa9, 0x19, 0xf0, 0xc1, 0x35, 0x64, 0xe7, 0xc8, 0x91, 0x7c, 0xe7, 0x39, 0x17,
+	0x78, 0xfe, 0xf4, 0x7e, 0x1d, 0x4e, 0x0c, 0xdf, 0x03, 0x08, 0x0d, 0xbf, 0x72, 0x48, 0xb8, 0x5a,
+	0x1c, 0x4e, 0x05, 0x91, 0xba, 0x43, 0xac, 0x83, 0x1f, 0x09, 0xc8, 0xce, 0x0c, 0xd8, 0x1b, 0x38,
+	0xb6, 0x47, 0xd1, 0x08, 0xb2, 0x21, 0x88, 0x78, 0x5e, 0xaf, 0x6b, 0xf7, 0xa9, 0xcd, 0xfc, 0x36,
+	0x7c, 0xc3, 0x94, 0x7b, 0x19, 0x16, 0xf2, 0x85, 0x8d, 0x89, 0x53, 0xae, 0xd0, 0x39, 0xde, 0x8b,
+	0x85, 0x51, 0x0b, 0x32, 0xfd, 0x5e, 0xd7, 0x25, 0xac, 0xe7, 0xd8, 0x66, 0xdf, 0xb1, 0xc2, 0x59,
+	0x65, 0x2a, 0xa5, 0xe5, 0xa2, 0x8d, 0x09, 0xae, 0xe1, 0x58, 0x14, 0xaf, 0xf7, 0x67, 0x5f, 0xf3,
+	0x23, 0xd8, 0x5c, 0xd8, 0xc2, 0x02, 0x97, 0xaa, 0xd1, 0xbd, 0x28, 0xff, 0xe1, 0xd4, 0xa7, 0xcc,
+	0xb3, 0xc6, 0xbe, 0x83, 0x7f, 0x62, 0x59, 0x74, 0x16, 0xb3, 0xb5, 0xb2, 0x5c, 0x60, 0x8a, 0x8e,
+	0x7a, 0x5b, 0xfc, 0xc2, 0x41, 0x26, 0x7a, 0xce, 0xd1, 0x0e, 0x6c, 0xcb, 0xaf, 0xe4, 0xd3, 0xa6,
+	0xa1, 0x61, 0x53, 0x37, 0x44, 0xa3, 0xa9, 0x9b, 0x8a, 0xda, 0x12, 0xeb, 0x8a, 0xc4, 0xff, 0x85,
+	0xf2, 0xb0, 0x15, 0x4f, 0x8a, 0xa7, 0x86, 0xd2, 0x92, 0x79, 0x0e, 0xed, 0x42, 0x2e, 0x9e, 0x93,
+	0xb0, 0xa8, 0xa8, 0x8a, 0x5a, 0xe5, 0x13, 0x8b, 0x68, 0x83, 0xac, 0x2c, 0xf1, 0xc9, 0x62, 0x0b,
+	0xd2, 0x33, 0x9b, 0x87, 0x72, 0xb0, 0xa1, 0xd7, 0x44, 0x2c, 0xcd, 0xeb, 0x6f, 0x01, 0x8a, 0x64,
+	0xb0, 0x2c, 0x4a, 0xaf, 0x79, 0x0e, 0x6d, 0x42, 0x36, 0x12, 0x97, 0x34, 0x55, 0xe6, 0x13, 0x45,
+	0x15, 0xf8, 0xf8, 0xa7, 0xa3, 0x3d, 0xf8, 0x57, 0xd4, 0x75, 0xa5, 0xaa, 0x36, 0x64, 0xd5, 0x98,
+	0x57, 0xd8, 0x81, 0xed, 0xf9, 0xf4, 0x58, 0xa6, 0x78, 0xc3, 0xc1, 0x7a, 0x64, 0x4d, 0xfc, 0x81,
+	0x34, 0x94, 0x2a, 0x16, 0x0d, 0x45, 0x53, 0xcd, 0x86, 0x26, 0xc9, 0x33, 0x54, 0xff, 0x43, 0x21,
+	0x96, 0xab, 0x6b, 0xa7, 0x62, 0xdd, 0x3c, 0x17, 0x75, 0xdd, 0xa8, 0x61, 0xad, 0x59, 0xad, 0xf1,
+	0x1c, 0x7a, 0x02, 0x8f, 0x97, 0x55, 0x99, 0x7a, 0x4d, 0x94, 0xb4, 0x0b, 0x3e, 0x81, 0x8a, 0xf0,
+	0x28, 0x56, 0x2c, 0x29, 0xba, 0x81, 0x95, 0x93, 0xa6, 0x21, 0x4b, 0x11, 0xe2, 0xa4, 0xef, 0x47,
+	0xac, 0x56, 0x53, 0x4f, 0x34, 0x11, 0x4b, 0xb2, 0xc4, 0xaf, 0x54, 0xbe, 0x72, 0xb0, 0x13, 0xcc,
+	0x5c, 0x9a, 0xae, 0xca, 0x64, 0x13, 0xc4, 0x73, 0x05, 0x31, 0x48, 0xdd, 0x9d, 0x42, 0x54, 0xb9,
+	0xff, 0x3f, 0x36, 0x7f, 0xf4, 0x80, 0x63, 0x7e, 0x72, 0x71, 0x73, 0xbb, 0xcf, 0x7d, 0xbf, 0xdd,
+	0xe7, 0x7e, 0xde, 0xee, 0x73, 0x6f, 0x94, 0x6e, 0x8f, 0x5d, 0x0e, 0xdb, 0x42, 0xc7, 0xe9, 0x47,
+	0xaf, 0x59, 0xa1, 0x4b, 0xed, 0x52, 0x70, 0x9d, 0x2e, 0xba, 0x71, 0x8f, 0xe3, 0xb1, 0x51, 0xb9,
+	0xbd, 0x16, 0x54, 0x1f, 0xfd, 0x0a, 0x00, 0x00, 0xff, 0xff, 0x33, 0x53, 0x4b, 0xe1, 0xaf, 0x07,
+	0x00, 0x00,
 }
 
 func (m *HeartbeatRequest) Marshal() (dAtA []byte, err error) {
@@ -469,6 +481,25 @@ func (m *HeartbeatRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
+	if len(m.Metadata) > 0 {
+		for k := range m.Metadata {
+			v := m.Metadata[k]
+			baseI := i
+			i -= len(v)
+			copy(dAtA[i:], v)
+			i = encodeVarintExecutor(dAtA, i, uint64(len(v)))
+			i--
+			dAtA[i] = 0x12
+			i -= len(k)
+			copy(dAtA[i:], k)
+			i = encodeVarintExecutor(dAtA, i, uint64(len(k)))
+			i--
+			dAtA[i] = 0xa
+			i = encodeVarintExecutor(dAtA, i, uint64(baseI-i))
+			i--
+			dAtA[i] = 0x2a
+		}
+	}
 	if len(m.ShardStatusReports) > 0 {
 		for k := range m.ShardStatusReports {
 			v := m.ShardStatusReports[k]
@@ -686,6 +717,14 @@ func (m *HeartbeatRequest) Size() (n int) {
 			n += mapEntrySize + 1 + sovExecutor(uint64(mapEntrySize))
 		}
 	}
+	if len(m.Metadata) > 0 {
+		for k, v := range m.Metadata {
+			_ = k
+			_ = v
+			mapEntrySize := 1 + len(k) + sovExecutor(uint64(len(k))) + 1 + len(v) + sovExecutor(uint64(len(v)))
+			n += mapEntrySize + 1 + sovExecutor(uint64(mapEntrySize))
+		}
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -1000,6 +1039,133 @@ func (m *HeartbeatRequest) Unmarshal(dAtA []byte) error {
 			}
 			m.ShardStatusReports[mapkey] = mapvalue
 			iNdEx = postIndex
+		case 5:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Metadata", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowExecutor
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthExecutor
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthExecutor
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.Metadata == nil {
+				m.Metadata = make(map[string]string)
+			}
+			var mapkey string
+			var mapvalue string
+			for iNdEx < postIndex {
+				entryPreIndex := iNdEx
+				var wire uint64
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowExecutor
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					wire |= uint64(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				fieldNum := int32(wire >> 3)
+				if fieldNum == 1 {
+					var stringLenmapkey uint64
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowExecutor
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						stringLenmapkey |= uint64(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					intStringLenmapkey := int(stringLenmapkey)
+					if intStringLenmapkey < 0 {
+						return ErrInvalidLengthExecutor
+					}
+					postStringIndexmapkey := iNdEx + intStringLenmapkey
+					if postStringIndexmapkey < 0 {
+						return ErrInvalidLengthExecutor
+					}
+					if postStringIndexmapkey > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapkey = string(dAtA[iNdEx:postStringIndexmapkey])
+					iNdEx = postStringIndexmapkey
+				} else if fieldNum == 2 {
+					var stringLenmapvalue uint64
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowExecutor
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						stringLenmapvalue |= uint64(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					intStringLenmapvalue := int(stringLenmapvalue)
+					if intStringLenmapvalue < 0 {
+						return ErrInvalidLengthExecutor
+					}
+					postStringIndexmapvalue := iNdEx + intStringLenmapvalue
+					if postStringIndexmapvalue < 0 {
+						return ErrInvalidLengthExecutor
+					}
+					if postStringIndexmapvalue > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapvalue = string(dAtA[iNdEx:postStringIndexmapvalue])
+					iNdEx = postStringIndexmapvalue
+				} else {
+					iNdEx = entryPreIndex
+					skippy, err := skipExecutor(dAtA[iNdEx:])
+					if err != nil {
+						return err
+					}
+					if (skippy < 0) || (iNdEx+skippy) < 0 {
+						return ErrInvalidLengthExecutor
+					}
+					if (iNdEx + skippy) > postIndex {
+						return io.ErrUnexpectedEOF
+					}
+					iNdEx += skippy
+				}
+			}
+			m.Metadata[mapkey] = mapvalue
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipExecutor(dAtA[iNdEx:])
diff --git a/.gen/proto/sharddistributor/v1/executor.pb.yarpc.go b/.gen/proto/sharddistributor/v1/executor.pb.yarpc.go
index 68925f8d0a0..fc5c022e743 100644
--- a/.gen/proto/sharddistributor/v1/executor.pb.yarpc.go
+++ b/.gen/proto/sharddistributor/v1/executor.pb.yarpc.go
@@ -233,50 +233,53 @@ var (
 var yarpcFileDescriptorClosure5aab034437d08cca = [][]byte{
 	// uber/cadence/sharddistributor/v1/executor.proto
 	[]byte{
-		0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x55, 0xdb, 0x6e, 0xda, 0x4a,
-		0x14, 0x3d, 0x86, 0x9c, 0x48, 0x6c, 0x14, 0x8e, 0x19, 0xe5, 0xc2, 0x21, 0x89, 0x8a, 0xa2, 0xaa,
-		0x8d, 0xa8, 0x6a, 0x0a, 0x79, 0xa9, 0xda, 0xa7, 0x49, 0x6c, 0xc1, 0xa4, 0x60, 0x47, 0x63, 0x43,
-		0x2f, 0x52, 0x65, 0x19, 0x3c, 0x22, 0x56, 0x83, 0x4d, 0xed, 0x01, 0x35, 0x55, 0x1f, 0xfb, 0x07,
-		0x55, 0xff, 0xa9, 0xff, 0xd0, 0x9f, 0xa9, 0x6c, 0x43, 0xc0, 0x86, 0x8a, 0xa6, 0x6f, 0xf6, 0x5e,
-		0x7b, 0xad, 0x3d, 0x5e, 0x6b, 0xc6, 0x03, 0xb5, 0x49, 0x9f, 0xf9, 0xb5, 0x81, 0x65, 0x33, 0x77,
-		0xc0, 0x6a, 0xc1, 0xb5, 0xe5, 0xdb, 0xb6, 0x13, 0x70, 0xdf, 0xe9, 0x4f, 0xb8, 0xe7, 0xd7, 0xa6,
-		0xf5, 0x1a, 0xfb, 0xc4, 0x06, 0xe1, 0xb3, 0x34, 0xf6, 0x3d, 0xee, 0xa1, 0x4a, 0x48, 0x90, 0x66,
-		0x04, 0x29, 0x4d, 0x90, 0xa6, 0xf5, 0x93, 0xef, 0x59, 0x10, 0x5b, 0xcc, 0xf2, 0x79, 0x9f, 0x59,
-		0x9c, 0xb2, 0x8f, 0x13, 0x16, 0x70, 0x74, 0x04, 0x39, 0xd7, 0x1a, 0xb1, 0x60, 0x6c, 0x0d, 0x58,
-		0x49, 0xa8, 0x08, 0xa7, 0x39, 0xba, 0x28, 0xa0, 0x07, 0x90, 0x9f, 0x8f, 0x31, 0x1d, 0xbb, 0x94,
-		0x89, 0x70, 0x98, 0x97, 0x88, 0x8d, 0x5a, 0xb0, 0x1d, 0x70, 0x8b, 0x4f, 0x82, 0x52, 0xb6, 0x22,
-		0x9c, 0x16, 0x1a, 0xcf, 0xa4, 0x4d, 0xcb, 0x90, 0x94, 0x19, 0x5b, 0x8f, 0x78, 0x74, 0xc6, 0x47,
-		0x5f, 0x60, 0x37, 0xea, 0x36, 0xe3, 0x77, 0xd3, 0x67, 0x63, 0xcf, 0xe7, 0x41, 0x69, 0xab, 0x92,
-		0x3d, 0xcd, 0x37, 0x2e, 0x37, 0xeb, 0xa6, 0x3f, 0x4d, 0xd2, 0xc3, 0xa6, 0xd9, 0x94, 0x58, 0x4c,
-		0x71, 0xb9, 0x7f, 0x4b, 0x51, 0xb0, 0x02, 0x94, 0x3f, 0xc3, 0xc1, 0x6f, 0xda, 0x91, 0x08, 0xd9,
-		0x0f, 0xec, 0x76, 0xe6, 0x4d, 0xf8, 0x88, 0x08, 0xfc, 0x3b, 0xb5, 0x6e, 0x26, 0x2c, 0xf2, 0x23,
-		0xdf, 0x38, 0xdb, 0xbc, 0xb6, 0x15, 0x6d, 0x1a, 0x2b, 0xbc, 0xc8, 0x3c, 0x17, 0x4e, 0x6e, 0xa1,
-		0xb8, 0x82, 0x23, 0xe5, 0xce, 0x58, 0x21, 0x32, 0xf6, 0xe9, 0xfd, 0x86, 0xcc, 0x5d, 0x3d, 0x06,
-		0x88, 0x5d, 0xbd, 0xf1, 0xac, 0x38, 0x3f, 0x81, 0xe6, 0xa2, 0x4a, 0xdb, 0xb3, 0xec, 0x93, 0x9f,
-		0x19, 0x28, 0x2e, 0xf9, 0x16, 0x8c, 0x3d, 0x37, 0x60, 0x68, 0x0a, 0xc5, 0x98, 0x64, 0x05, 0x81,
-		0x33, 0x74, 0x47, 0xcc, 0xe5, 0xe1, 0x32, 0xc2, 0x1c, 0xc8, 0xbd, 0x72, 0x88, 0xf5, 0xe2, 0x85,
-		0xe1, 0x85, 0x56, 0x1c, 0x83, 0x18, 0xa4, 0xca, 0xa8, 0x07, 0x85, 0x91, 0x33, 0xf4, 0x2d, 0xee,
-		0x78, 0xae, 0x39, 0xf2, 0xec, 0xd8, 0xe0, 0x42, 0xa3, 0xb6, 0x79, 0x68, 0x67, 0xce, 0xeb, 0x78,
-		0x36, 0xa3, 0x3b, 0xa3, 0xe5, 0xd7, 0xf2, 0x14, 0xf6, 0xd6, 0x2e, 0x61, 0x4d, 0xb4, 0xcd, 0x64,
-		0xb4, 0xf5, 0x3f, 0x74, 0x7d, 0xa1, 0xbc, 0x1c, 0xec, 0x7b, 0xf8, 0x2f, 0x85, 0xa2, 0xcb, 0x54,
-		0xac, 0x8d, 0xcd, 0x03, 0x16, 0xec, 0x64, 0xb6, 0xd5, 0xaf, 0x02, 0x14, 0x92, 0x87, 0x09, 0x1d,
-		0xc2, 0x81, 0xf2, 0x46, 0xb9, 0xe8, 0x1a, 0x1a, 0x35, 0x75, 0x03, 0x1b, 0x5d, 0xdd, 0x24, 0x6a,
-		0x0f, 0xb7, 0x89, 0x2c, 0xfe, 0x83, 0xca, 0xb0, 0x9f, 0x06, 0xf1, 0x85, 0x41, 0x7a, 0x8a, 0x28,
-		0xa0, 0x23, 0x28, 0xa5, 0x31, 0x99, 0x62, 0xa2, 0x12, 0xb5, 0x29, 0x66, 0xd6, 0xc9, 0x46, 0xa8,
-		0x22, 0x8b, 0xd9, 0x6a, 0x0f, 0xf2, 0x4b, 0x3b, 0x0f, 0x95, 0x60, 0x57, 0x6f, 0x61, 0x2a, 0xaf,
-		0xce, 0xdf, 0x07, 0x94, 0x40, 0xa8, 0x82, 0xe5, 0xb7, 0xa2, 0x80, 0xf6, 0xa0, 0x98, 0xa8, 0xcb,
-		0x9a, 0xaa, 0x88, 0x99, 0xaa, 0x0a, 0x62, 0xfa, 0xd3, 0xd1, 0x31, 0xfc, 0x8f, 0x75, 0x9d, 0x34,
-		0xd5, 0x8e, 0xa2, 0x1a, 0xab, 0x13, 0x0e, 0xe1, 0x60, 0x15, 0x9e, 0x8d, 0xa9, 0xfe, 0x10, 0x60,
-		0x27, 0xb1, 0x4d, 0x42, 0x43, 0x3a, 0xa4, 0x49, 0xb1, 0x41, 0x34, 0xd5, 0xec, 0x68, 0xb2, 0xb2,
-		0x24, 0xf5, 0x10, 0x2a, 0x29, 0xac, 0xad, 0x5d, 0xe0, 0xb6, 0x79, 0x85, 0x75, 0xdd, 0x68, 0x51,
-		0xad, 0xdb, 0x6c, 0x89, 0x02, 0x7a, 0x02, 0x8f, 0x37, 0x75, 0x99, 0x7a, 0x0b, 0xcb, 0xda, 0x6b,
-		0x31, 0x83, 0xaa, 0xf0, 0x28, 0xd5, 0x2c, 0x13, 0xdd, 0xa0, 0xe4, 0xbc, 0x6b, 0x28, 0x72, 0x42,
-		0x38, 0x1b, 0xe6, 0x91, 0xea, 0xd5, 0xd4, 0x73, 0x0d, 0x53, 0x59, 0x91, 0xc5, 0xad, 0xc6, 0x37,
-		0x01, 0x0e, 0x23, 0xcf, 0xe5, 0xc5, 0x56, 0x99, 0xef, 0x04, 0x7c, 0x45, 0x10, 0x87, 0xdc, 0xdd,
-		0x29, 0x44, 0x8d, 0xfb, 0xff, 0x3a, 0xcb, 0x67, 0x7f, 0x71, 0xcc, 0xcf, 0x5f, 0xbd, 0x23, 0x43,
-		0x87, 0x5f, 0x4f, 0xfa, 0xd2, 0xc0, 0x1b, 0x25, 0xef, 0x2f, 0x69, 0xc8, 0xdc, 0x5a, 0x74, 0x4f,
-		0xad, 0xbb, 0xca, 0x5e, 0xa6, 0x6b, 0xd3, 0x7a, 0x7f, 0x3b, 0xea, 0x3e, 0xfb, 0x15, 0x00, 0x00,
-		0xff, 0xff, 0xa7, 0x93, 0xad, 0x22, 0x08, 0x07, 0x00, 0x00,
+		0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x55, 0xdd, 0x6e, 0xda, 0x48,
+		0x14, 0x5e, 0x43, 0x12, 0x2d, 0x07, 0x85, 0x35, 0xa3, 0xfc, 0xb0, 0x24, 0xd1, 0xa2, 0x68, 0xb5,
+		0x1b, 0x51, 0xd5, 0x14, 0x72, 0x53, 0x35, 0x37, 0x75, 0x62, 0x0b, 0x26, 0x05, 0x3b, 0x1a, 0x1b,
+		0xfa, 0xa3, 0x56, 0xd6, 0x80, 0x47, 0x04, 0x35, 0xd8, 0xd4, 0x1e, 0x50, 0x53, 0xf5, 0xb2, 0x6f,
+		0xd0, 0x17, 0xe9, 0x63, 0xf4, 0x1d, 0xfa, 0x32, 0x95, 0x6d, 0x08, 0xd8, 0x50, 0xd1, 0xe4, 0xce,
+		0x3e, 0xe7, 0x7c, 0xdf, 0x77, 0x7c, 0xbe, 0x33, 0x1e, 0xa8, 0x8c, 0xbb, 0xcc, 0xab, 0xf4, 0xa8,
+		0xcd, 0x9c, 0x1e, 0xab, 0xf8, 0xd7, 0xd4, 0xb3, 0xed, 0x81, 0xcf, 0xbd, 0x41, 0x77, 0xcc, 0x5d,
+		0xaf, 0x32, 0xa9, 0x56, 0xd8, 0x47, 0xd6, 0x0b, 0x9e, 0xa5, 0x91, 0xe7, 0x72, 0x17, 0x95, 0x02,
+		0x80, 0x34, 0x05, 0x48, 0x49, 0x80, 0x34, 0xa9, 0x1e, 0x7f, 0xdb, 0x00, 0xb1, 0xc1, 0xa8, 0xc7,
+		0xbb, 0x8c, 0x72, 0xc2, 0x3e, 0x8c, 0x99, 0xcf, 0xd1, 0x21, 0x64, 0x1c, 0x3a, 0x64, 0xfe, 0x88,
+		0xf6, 0x58, 0x41, 0x28, 0x09, 0x27, 0x19, 0x32, 0x0f, 0xa0, 0x7f, 0x20, 0x3b, 0x93, 0xb1, 0x06,
+		0x76, 0x21, 0x15, 0xe6, 0x61, 0x16, 0xc2, 0x36, 0x6a, 0xc0, 0x96, 0xcf, 0x29, 0x1f, 0xfb, 0x85,
+		0x74, 0x49, 0x38, 0xc9, 0xd5, 0x9e, 0x48, 0xeb, 0xda, 0x90, 0xd4, 0x29, 0xda, 0x08, 0x71, 0x64,
+		0x8a, 0x47, 0x9f, 0x61, 0x27, 0xac, 0xb6, 0xa2, 0x77, 0xcb, 0x63, 0x23, 0xd7, 0xe3, 0x7e, 0x61,
+		0xa3, 0x94, 0x3e, 0xc9, 0xd6, 0x2e, 0xd7, 0xf3, 0x26, 0x3f, 0x4d, 0x32, 0x82, 0xa2, 0xa9, 0x4a,
+		0x44, 0xa6, 0x3a, 0xdc, 0xbb, 0x25, 0xc8, 0x5f, 0x4a, 0xa0, 0xb7, 0xf0, 0xe7, 0x90, 0x71, 0x6a,
+		0x53, 0x4e, 0x0b, 0x9b, 0xa1, 0xe2, 0xf3, 0x07, 0x28, 0xb6, 0xa6, 0x14, 0x91, 0xce, 0x1d, 0x63,
+		0xf1, 0x13, 0xec, 0xff, 0xa2, 0x19, 0x24, 0x42, 0xfa, 0x3d, 0xbb, 0x9d, 0x4e, 0x3e, 0x78, 0x44,
+		0x18, 0x36, 0x27, 0xf4, 0x66, 0xcc, 0xc2, 0x69, 0x67, 0x6b, 0xa7, 0xeb, 0xfb, 0x58, 0xe2, 0x26,
+		0x11, 0xc3, 0xb3, 0xd4, 0x53, 0xa1, 0x78, 0x06, 0xdb, 0xb1, 0xb6, 0x56, 0x28, 0xee, 0x2c, 0x2a,
+		0x66, 0x16, 0xc0, 0xc7, 0xb7, 0x90, 0x5f, 0x22, 0x47, 0xea, 0x9d, 0xe7, 0x42, 0xe8, 0xf9, 0xe3,
+		0xfb, 0x75, 0x38, 0x33, 0xfc, 0x08, 0x20, 0x32, 0xfc, 0xc6, 0xa5, 0xd1, 0x6a, 0x09, 0x24, 0x13,
+		0x46, 0x9a, 0x2e, 0xb5, 0x8f, 0x7f, 0xa4, 0x20, 0xbf, 0x30, 0x60, 0x7f, 0xe4, 0x3a, 0x3e, 0x43,
+		0x13, 0xc8, 0x47, 0x20, 0xea, 0xfb, 0x83, 0xbe, 0x33, 0x64, 0x0e, 0x0f, 0xda, 0x08, 0x0c, 0xc3,
+		0xf7, 0x32, 0x2c, 0xe2, 0x8b, 0x1a, 0x93, 0xe7, 0x5c, 0x91, 0x73, 0xa2, 0x9f, 0x08, 0xa3, 0x0e,
+		0xe4, 0x86, 0x83, 0xbe, 0x47, 0xf9, 0xc0, 0x75, 0xac, 0xa1, 0x6b, 0x47, 0xb3, 0xca, 0xd5, 0x2a,
+		0xeb, 0x45, 0x5b, 0x33, 0x5c, 0xcb, 0xb5, 0x19, 0xd9, 0x1e, 0x2e, 0xbe, 0x16, 0x27, 0xb0, 0xbb,
+		0xb2, 0x85, 0x15, 0x2e, 0xd5, 0xe3, 0x7b, 0x51, 0xfd, 0xcd, 0xa9, 0xcf, 0x99, 0x17, 0x8d, 0x7d,
+		0x07, 0x7f, 0x25, 0xb2, 0xe8, 0x32, 0x61, 0x6b, 0x6d, 0xbd, 0xc0, 0x1c, 0x1d, 0xf7, 0xb6, 0xfc,
+		0x45, 0x80, 0x5c, 0xfc, 0x9c, 0xa3, 0x03, 0xd8, 0x57, 0x5f, 0xa9, 0x17, 0x6d, 0x53, 0x27, 0x96,
+		0x61, 0xca, 0x66, 0xdb, 0xb0, 0xb0, 0xd6, 0x91, 0x9b, 0x58, 0x11, 0xff, 0x40, 0x45, 0xd8, 0x4b,
+		0x26, 0xe5, 0x0b, 0x13, 0x77, 0x54, 0x51, 0x40, 0x87, 0x50, 0x48, 0xe6, 0x14, 0x22, 0x63, 0x0d,
+		0x6b, 0x75, 0x31, 0xb5, 0x8a, 0x36, 0xcc, 0xaa, 0x8a, 0x98, 0x2e, 0x77, 0x20, 0xbb, 0xb0, 0x79,
+		0xa8, 0x00, 0x3b, 0x46, 0x43, 0x26, 0xca, 0xb2, 0xfe, 0x1e, 0xa0, 0x58, 0x86, 0xa8, 0xb2, 0xf2,
+		0x5a, 0x14, 0xd0, 0x2e, 0xe4, 0x63, 0x71, 0x45, 0xd7, 0x54, 0x31, 0x55, 0xd6, 0x40, 0x4c, 0x7e,
+		0x3a, 0x3a, 0x82, 0xbf, 0x65, 0xc3, 0xc0, 0x75, 0xad, 0xa5, 0x6a, 0xe6, 0xb2, 0xc2, 0x01, 0xec,
+		0x2f, 0xa7, 0xa7, 0x32, 0xe5, 0xef, 0x02, 0x6c, 0xc7, 0xd6, 0x24, 0x18, 0x48, 0x0b, 0xd7, 0x89,
+		0x6c, 0x62, 0x5d, 0xb3, 0x5a, 0xba, 0xa2, 0x2e, 0x50, 0xfd, 0x0b, 0xa5, 0x44, 0xae, 0xa9, 0x5f,
+		0xc8, 0x4d, 0xeb, 0x4a, 0x36, 0x0c, 0xb3, 0x41, 0xf4, 0x76, 0xbd, 0x21, 0x0a, 0xe8, 0x11, 0xfc,
+		0xbf, 0xae, 0xca, 0x32, 0x1a, 0xb2, 0xa2, 0xbf, 0x14, 0x53, 0xa8, 0x0c, 0xff, 0x25, 0x8a, 0x15,
+		0x6c, 0x98, 0x04, 0x9f, 0xb7, 0x4d, 0x55, 0x89, 0x11, 0xa7, 0x03, 0x3f, 0x12, 0xb5, 0xba, 0x76,
+		0xae, 0xcb, 0x44, 0x51, 0x15, 0x71, 0xa3, 0xf6, 0x55, 0x80, 0x83, 0x70, 0xe6, 0xca, 0x7c, 0x55,
+		0x66, 0x9b, 0x20, 0x5f, 0x61, 0xc4, 0x21, 0x73, 0x77, 0x0a, 0x51, 0xed, 0xfe, 0xff, 0xd8, 0xe2,
+		0xe9, 0x03, 0x8e, 0xf9, 0xf9, 0x8b, 0x37, 0xb8, 0x3f, 0xe0, 0xd7, 0xe3, 0xae, 0xd4, 0x73, 0x87,
+		0xf1, 0xab, 0x55, 0xea, 0x33, 0xa7, 0x12, 0x5e, 0xa1, 0xab, 0x6e, 0xd9, 0xb3, 0x64, 0x6c, 0x52,
+		0xed, 0x6e, 0x85, 0xd5, 0xa7, 0x3f, 0x03, 0x00, 0x00, 0xff, 0xff, 0x7a, 0x3d, 0x1d, 0x21, 0xa3,
+		0x07, 0x00, 0x00,
 	},
 }
 
diff --git a/common/types/mapper/proto/sharddistributor.go b/common/types/mapper/proto/sharddistributor.go
index 92a99c2c8db..0cb8d65208c 100644
--- a/common/types/mapper/proto/sharddistributor.go
+++ b/common/types/mapper/proto/sharddistributor.go
@@ -121,6 +121,7 @@ func FromShardDistributorExecutorHeartbeatRequest(t *types.ExecutorHeartbeatRequ
 		ExecutorId:         t.GetExecutorID(),
 		Status:             status,
 		ShardStatusReports: shardStatusReports,
+		Metadata:           t.GetMetadata(),
 	}
 }
 
@@ -173,6 +174,7 @@ func ToShardDistributorExecutorHeartbeatRequest(t *sharddistributorv1.HeartbeatR
 		ExecutorID:         t.GetExecutorId(),
 		Status:             status,
 		ShardStatusReports: shardStatusReports,
+		Metadata:           t.GetMetadata(),
 	}
 }
 
diff --git a/common/types/sharddistributor.go b/common/types/sharddistributor.go
index 97c5eab7f30..cd682c44e32 100644
--- a/common/types/sharddistributor.go
+++ b/common/types/sharddistributor.go
@@ -92,6 +92,7 @@ type ExecutorHeartbeatRequest struct {
 	ExecutorID         string
 	Status             ExecutorStatus
 	ShardStatusReports map[string]*ShardStatusReport
+	Metadata           map[string]string
 }
 
 func (v *ExecutorHeartbeatRequest) GetNamespace() (o string) {
@@ -122,6 +123,13 @@ func (v *ExecutorHeartbeatRequest) GetShardStatusReports() (o map[string]*ShardS
 	return
 }
 
+func (v *ExecutorHeartbeatRequest) GetMetadata() (o map[string]string) {
+	if v != nil {
+		return v.Metadata
+	}
+	return
+}
+
 // ExecutorStatus is persisted to the DB with a string value mapping.
 // Beware - if we want to change the name - it should be backward compatible and should be done in two steps.
 type ExecutorStatus int32
diff --git a/common/types/testdata/service_sharddistributor.go b/common/types/testdata/service_sharddistributor.go
index dd18f24d8e2..683f5289d08 100644
--- a/common/types/testdata/service_sharddistributor.go
+++ b/common/types/testdata/service_sharddistributor.go
@@ -47,6 +47,10 @@ var (
 				ShardLoad: 0.75,
 			},
 		},
+		Metadata: map[string]string{
+			"key-1": "value-1",
+			"key-2": "value-2",
+		},
 	}
 	ShardDistributorExecutorHeartbeatResponse = types.ExecutorHeartbeatResponse{
 		ShardAssignments: map[string]*types.ShardAssignment{
diff --git a/proto/internal/uber/cadence/sharddistributor/v1/executor.proto b/proto/internal/uber/cadence/sharddistributor/v1/executor.proto
index 83a2fc9a781..6adcdd2a440 100644
--- a/proto/internal/uber/cadence/sharddistributor/v1/executor.proto
+++ b/proto/internal/uber/cadence/sharddistributor/v1/executor.proto
@@ -36,6 +36,7 @@ message HeartbeatRequest {
   string executor_id = 2;
   ExecutorStatus status = 3;
   map<string, ShardStatusReport> shard_status_reports = 4;
+  map<string, string> metadata = 5;
 }
 
 enum ExecutorStatus {
diff --git a/service/sharddistributor/handler/executor.go b/service/sharddistributor/handler/executor.go
index 2a5e12f5c43..db1c22cfb00 100644
--- a/service/sharddistributor/handler/executor.go
+++ b/service/sharddistributor/handler/executor.go
@@ -16,6 +16,10 @@ import (
 
 const (
 	_heartbeatRefreshRate = 2 * time.Second
+
+	_maxMetadataKeys      = 32
+	_maxMetadataKeyLength = 128
+	_maxMetadataValueSize = 512 * 1024 // 512KB
 )
 
 type executor struct {
@@ -78,6 +82,11 @@ func (h *executor) Heartbeat(ctx context.Context, request *types.ExecutorHeartbe
 		LastHeartbeat:  now.Unix(),
 		Status:         request.Status,
 		ReportedShards: request.ShardStatusReports,
+		Metadata:       request.GetMetadata(),
+	}
+
+	if err := validateMetadata(newHeartbeat.Metadata); err != nil {
+		return nil, fmt.Errorf("validate metadata: %w", err)
 	}
 
 	err = h.storage.RecordHeartbeat(ctx, request.Namespace, request.ExecutorID, newHeartbeat)
@@ -130,3 +139,21 @@ func _convertResponse(shards *store.AssignedState, mode types.MigrationMode) *ty
 	res.MigrationMode = mode
 	return res
 }
+
+func validateMetadata(metadata map[string]string) error {
+	if len(metadata) > _maxMetadataKeys {
+		return fmt.Errorf("metadata has %d keys, which exceeds the maximum of %d", len(metadata), _maxMetadataKeys)
+	}
+
+	for key, value := range metadata {
+		if len(key) > _maxMetadataKeyLength {
+			return fmt.Errorf("metadata key %q has length %d, which exceeds the maximum of %d", key, len(key), _maxMetadataKeyLength)
+		}
+
+		if len(value) > _maxMetadataValueSize {
+			return fmt.Errorf("metadata value for key %q has size %d bytes, which exceeds the maximum of %d bytes", key, len(value), _maxMetadataValueSize)
+		}
+	}
+
+	return nil
+}
diff --git a/service/sharddistributor/handler/executor_test.go b/service/sharddistributor/handler/executor_test.go
index 42bfc52782b..fbc9508e258 100644
--- a/service/sharddistributor/handler/executor_test.go
+++ b/service/sharddistributor/handler/executor_test.go
@@ -356,6 +356,136 @@ func TestHeartbeat(t *testing.T) {
 		require.Error(t, err)
 		require.Contains(t, err.Error(), expectedErr.Error())
 	})
+
+	// Test Case 11: Heartbeat with metadata validation failure - too many keys
+	t.Run("MetadataValidationTooManyKeys", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		mockStore := store.NewMockStore(ctrl)
+		mockTimeSource := clock.NewMockedTimeSourceAt(now)
+		shardDistributionCfg := config.ShardDistribution{}
+		handler := NewExecutorHandler(testlogger.New(t), mockStore, mockTimeSource, shardDistributionCfg)
+
+		// Create metadata with more than max allowed keys
+		metadata := make(map[string]string)
+		for i := 0; i < _maxMetadataKeys+1; i++ {
+			metadata[string(rune('a'+i))] = "value"
+		}
+
+		req := &types.ExecutorHeartbeatRequest{
+			Namespace:  namespace,
+			ExecutorID: executorID,
+			Status:     types.ExecutorStatusACTIVE,
+			Metadata:   metadata,
+		}
+
+		mockStore.EXPECT().GetHeartbeat(gomock.Any(), namespace, executorID).Return(nil, nil, store.ErrExecutorNotFound)
+
+		_, err := handler.Heartbeat(ctx, req)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "validate metadata")
+		require.Contains(t, err.Error(), "exceeds the maximum")
+	})
+}
+
+func TestValidateMetadata(t *testing.T) {
+	// Helper function to generate metadata with N keys
+	makeMetadataWithKeys := func(n int) map[string]string {
+		metadata := make(map[string]string)
+		for i := 0; i < n; i++ {
+			metadata[string(rune('a'+i))] = "value"
+		}
+		return metadata
+	}
+
+	testCases := []struct {
+		name           string
+		metadata       map[string]string
+		expectError    bool
+		errorSubstring string
+	}{
+		{
+			name: "ValidMetadata",
+			metadata: map[string]string{
+				"key1": "value1",
+				"key2": "value2",
+			},
+			expectError: false,
+		},
+		{
+			name:        "EmptyMetadata",
+			metadata:    map[string]string{},
+			expectError: false,
+		},
+		{
+			name:        "NilMetadata",
+			metadata:    nil,
+			expectError: false,
+		},
+		{
+			name:           "TooManyKeys",
+			metadata:       makeMetadataWithKeys(_maxMetadataKeys + 1),
+			expectError:    true,
+			errorSubstring: "exceeds the maximum of 32",
+		},
+		{
+			name:        "ExactlyMaxKeys",
+			metadata:    makeMetadataWithKeys(_maxMetadataKeys),
+			expectError: false,
+		},
+		{
+			name: "KeyTooLong",
+			metadata: map[string]string{
+				string(make([]byte, _maxMetadataKeyLength+1)): "value",
+			},
+			expectError:    true,
+			errorSubstring: "exceeds the maximum of 128",
+		},
+		{
+			name: "KeyExactlyMaxLength",
+			metadata: map[string]string{
+				string(make([]byte, _maxMetadataKeyLength)): "value",
+			},
+			expectError: false,
+		},
+		{
+			name: "ValueTooLarge",
+			metadata: map[string]string{
+				"key": string(make([]byte, _maxMetadataValueSize+1)),
+			},
+			expectError:    true,
+			errorSubstring: "exceeds the maximum of 524288 bytes",
+		},
+		{
+			name: "ValueExactlyMaxSize",
+			metadata: map[string]string{
+				"key": string(make([]byte, _maxMetadataValueSize)),
+			},
+			expectError: false,
+		},
+		{
+			name: "MultipleValidationErrors",
+			metadata: func() map[string]string {
+				metadata := makeMetadataWithKeys(_maxMetadataKeys + 1)
+				longKey := string(make([]byte, _maxMetadataKeyLength+1))
+				metadata[longKey] = "value"
+				return metadata
+			}(),
+			expectError:    true,
+			errorSubstring: "exceeds the maximum of 32", // First validation error
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := validateMetadata(tc.metadata)
+			if tc.expectError {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tc.errorSubstring)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
 }
 
 func TestConvertResponse(t *testing.T) {
diff --git a/service/sharddistributor/store/etcd/etcdkeys/etcdkeys.go b/service/sharddistributor/store/etcd/etcdkeys/etcdkeys.go
index aad2638e35e..d9a1d35be9f 100644
--- a/service/sharddistributor/store/etcd/etcdkeys/etcdkeys.go
+++ b/service/sharddistributor/store/etcd/etcdkeys/etcdkeys.go
@@ -11,6 +11,7 @@ const (
 	ExecutorReportedShardsKey = "reported_shards"
 	ExecutorAssignedStateKey  = "assigned_state"
 	ShardAssignedKey          = "assigned"
+	ExecutorMetadataKey       = "metadata"
 )
 
 var validKeyTypes = []string{
@@ -18,6 +19,7 @@ var validKeyTypes = []string{
 	ExecutorStatusKey,
 	ExecutorReportedShardsKey,
 	ExecutorAssignedStateKey,
+	ExecutorMetadataKey,
 }
 
 func isValidKeyType(key string) bool {
@@ -57,3 +59,12 @@ func ParseExecutorKey(prefix string, namespace, key string) (executorID, keyType
 	}
 	return parts[0], parts[1], nil
 }
+
+func BuildMetadataKey(prefix string, namespace, executorID, metadataKey string) string {
+	metadataKeyPrefix, err := BuildExecutorKey(prefix, namespace, executorID, ExecutorMetadataKey)
+	if err != nil {
+		// This should never happen since ExecutorMetadataKey is a valid constant
+		panic(fmt.Sprintf("BuildMetadataKey: unexpected error building executor key: %v", err))
+	}
+	return fmt.Sprintf("%s/%s", metadataKeyPrefix, metadataKey)
+}
diff --git a/service/sharddistributor/store/etcd/etcdkeys/etcdkeys_test.go b/service/sharddistributor/store/etcd/etcdkeys/etcdkeys_test.go
index c98a01f2b57..81e2dd37393 100644
--- a/service/sharddistributor/store/etcd/etcdkeys/etcdkeys_test.go
+++ b/service/sharddistributor/store/etcd/etcdkeys/etcdkeys_test.go
@@ -42,3 +42,8 @@ func TestParseExecutorKey(t *testing.T) {
 	_, _, err = ParseExecutorKey("/cadence", "test-ns", "/cadence/test-ns/executors/exec-1/heartbeat/extra")
 	assert.ErrorContains(t, err, "unexpected key format: /cadence/test-ns/executors/exec-1/heartbeat/extra")
 }
+
+func TestBuildMetadataKey(t *testing.T) {
+	got := BuildMetadataKey("/cadence", "test-ns", "exec-1", "my-metadata-key")
+	assert.Equal(t, "/cadence/test-ns/executors/exec-1/metadata/my-metadata-key", got)
+}
diff --git a/service/sharddistributor/store/etcd/executorstore/etcdstore.go b/service/sharddistributor/store/etcd/executorstore/etcdstore.go
index e8fc6f50d73..700028fd747 100644
--- a/service/sharddistributor/store/etcd/executorstore/etcdstore.go
+++ b/service/sharddistributor/store/etcd/executorstore/etcdstore.go
@@ -121,12 +121,19 @@ func (s *executorStoreImpl) RecordHeartbeat(ctx context.Context, namespace, exec
 		return fmt.Errorf("marshal assinged shards: %w", err)
 	}
 
-	// Atomically update both the timestamp and the state.
-	_, err = s.client.Txn(ctx).Then(
+	// Build all operations including metadata
+	ops := []clientv3.Op{
 		clientv3.OpPut(heartbeatETCDKey, strconv.FormatInt(request.LastHeartbeat, 10)),
 		clientv3.OpPut(stateETCDKey, string(jsonState)),
 		clientv3.OpPut(reportedShardsETCDKey, string(reportedShardsData)),
-	).Commit()
+	}
+	for key, value := range request.Metadata {
+		metadataKey := etcdkeys.BuildMetadataKey(s.prefix, namespace, executorID, key)
+		ops = append(ops, clientv3.OpPut(metadataKey, value))
+	}
+
+	// Atomically update both the timestamp and the state.
+	_, err = s.client.Txn(ctx).Then(ops...).Commit()
 
 	if err != nil {
 		return fmt.Errorf("record heartbeat: %w", err)
diff --git a/service/sharddistributor/store/etcd/executorstore/etcdstore_test.go b/service/sharddistributor/store/etcd/executorstore/etcdstore_test.go
index ff3dc815e36..903693c00d5 100644
--- a/service/sharddistributor/store/etcd/executorstore/etcdstore_test.go
+++ b/service/sharddistributor/store/etcd/executorstore/etcdstore_test.go
@@ -36,6 +36,10 @@ func TestRecordHeartbeat(t *testing.T) {
 		ReportedShards: map[string]*types.ShardStatusReport{
 			"shard-TestRecordHeartbeat": {Status: types.ShardStatusREADY},
 		},
+		Metadata: map[string]string{
+			"key-1": "value-1",
+			"key-2": "value-2",
+		},
 	}
 
 	err := executorStore.RecordHeartbeat(ctx, tc.Namespace, executorID, req)
@@ -48,6 +52,8 @@ func TestRecordHeartbeat(t *testing.T) {
 	require.NoError(t, err)
 	reportedShardsKey, err := etcdkeys.BuildExecutorKey(tc.EtcdPrefix, tc.Namespace, executorID, etcdkeys.ExecutorReportedShardsKey)
 	require.NoError(t, err)
+	metadataKey1 := etcdkeys.BuildMetadataKey(tc.EtcdPrefix, tc.Namespace, executorID, "key-1")
+	metadataKey2 := etcdkeys.BuildMetadataKey(tc.EtcdPrefix, tc.Namespace, executorID, "key-2")
 
 	resp, err := tc.Client.Get(ctx, heartbeatKey)
 	require.NoError(t, err)
@@ -68,6 +74,16 @@ func TestRecordHeartbeat(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, reportedShards, 1)
 	assert.Equal(t, types.ShardStatusREADY, reportedShards["shard-TestRecordHeartbeat"].Status)
+
+	resp, err = tc.Client.Get(ctx, metadataKey1)
+	require.NoError(t, err)
+	require.Equal(t, int64(1), resp.Count, "Metadata key 1 should exist")
+	assert.Equal(t, "value-1", string(resp.Kvs[0].Value))
+
+	resp, err = tc.Client.Get(ctx, metadataKey2)
+	require.NoError(t, err)
+	require.Equal(t, int64(1), resp.Count, "Metadata key 2 should exist")
+	assert.Equal(t, "value-2", string(resp.Kvs[0].Value))
 }
 
 func TestGetHeartbeat(t *testing.T) {
diff --git a/service/sharddistributor/store/state.go b/service/sharddistributor/store/state.go
index 2020df0988d..d0d8315afc8 100644
--- a/service/sharddistributor/store/state.go
+++ b/service/sharddistributor/store/state.go
@@ -8,6 +8,7 @@ type HeartbeatState struct {
 	LastHeartbeat  int64                               `json:"last_heartbeat"`
 	Status         types.ExecutorStatus                `json:"status"`
 	ReportedShards map[string]*types.ShardStatusReport `json:"reported_shards"`
+	Metadata       map[string]string                   `json:"metadata"`
 }
 
 type AssignedState struct {