Skip to content
This repository has been archived by the owner on Feb 18, 2021. It is now read-only.

RetMgr: Handle cleanup of consumer-groups and consumer-group extents #310

Merged
merged 22 commits into from
Oct 20, 2017
Merged

RetMgr: Handle cleanup of consumer-groups and consumer-group extents #310

Show file tree
Hide file tree
Changes from 18 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
f833838
handle consumer-group deletion
Sep 30, 2017
7f0038b
fix mocks, tests
Oct 3, 2017
e8423e1
debug log
Oct 4, 2017
14c0877
more log
Oct 4, 2017
2f9b6ff
log
Oct 4, 2017
aefff4f
metadata logs
Oct 4, 2017
885b290
add destination_uuid column to consumer_groups table
Oct 10, 2017
a6ae039
ListConsumerGroupsUUID
Oct 10, 2017
b90267b
Make RetMgr use ListConsumerGroupsUUID
Oct 10, 2017
e125318
update meta-metrics
Oct 10, 2017
df02f11
thrift
Oct 10, 2017
9ee4008
thrift
Oct 10, 2017
ed2765f
update the new dest-uuid column in create-, update- and delete- cg paths
Oct 11, 2017
6325258
fix create-cg cql
Oct 13, 2017
e3d590f
fix mocks
Oct 17, 2017
c674b61
ListConsumerGroupsUUID test
Oct 18, 2017
3979198
CR feedback
Oct 18, 2017
33fc0c8
return and catch errors from metadata in ret-mgr
Oct 18, 2017
d36c008
incorporate cr feedback
Oct 19, 2017
6c70a9c
~
Oct 19, 2017
f63bfdf
ensure consumer-groups are deleted before deleting destination
Oct 20, 2017
1abe1d8
fix build due to gocql API change
Oct 20, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 102 additions & 48 deletions clients/metadata/metadata_cassandra.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1189,12 +1189,29 @@ const (
columnZoneConfigs + `: ?, ` +
columnOptions + `: ? }`

sqlConsumerGroupType = columnConsumerGroup + `.` + columnUUID + "," +
columnConsumerGroup + `.` + columnDestinationUUID + "," +
columnConsumerGroup + `.` + columnName + "," +
columnConsumerGroup + `.` + columnStartFrom + "," +
columnConsumerGroup + `.` + columnStatus + "," +
columnConsumerGroup + `.` + columnLockTimeoutSeconds + "," +
columnConsumerGroup + `.` + columnMaxDeliveryCount + "," +
columnConsumerGroup + `.` + columnSkipOlderMessagesSeconds + "," +
columnConsumerGroup + `.` + columnDelaySeconds + "," +
columnConsumerGroup + `.` + columnDeadLetterQueueDestinationUUID + "," +
columnConsumerGroup + `.` + columnOwnerEmail + "," +
columnConsumerGroup + `.` + columnIsMultiZone + "," +
columnConsumerGroup + `.` + columnActiveZone + "," +
columnConsumerGroup + `.` + columnZoneConfigs + "," +
columnConsumerGroup + `.` + columnOptions

sqlInsertCGByUUID = `INSERT INTO ` + tableConsumerGroups +
`(` +
columnUUID + `, ` +
columnDestinationUUID + `, ` +
columnIsMultiZone + `, ` +
columnConsumerGroup +
`) VALUES (?, ?,` + sqlCGValue + `)`
`) VALUES (?, ?, ?, ` + sqlCGValue + `)`

sqlInsertCGByName = `INSERT INTO ` + tableConsumerGroupsByName +
`(` +
Expand All @@ -1205,65 +1222,28 @@ const (
`) VALUES (?, ?, ?, ` + sqlCGValue + `) IF NOT EXISTS`

sqlGetCGByName = `SELECT ` +
columnConsumerGroup + `.` + columnUUID + "," +
columnDestinationUUID + "," +
columnName + "," +
columnConsumerGroup + `.` + columnStartFrom + "," +
columnConsumerGroup + `.` + columnStatus + "," +
columnConsumerGroup + `.` + columnLockTimeoutSeconds + "," +
columnConsumerGroup + `.` + columnMaxDeliveryCount + "," +
columnConsumerGroup + `.` + columnSkipOlderMessagesSeconds + "," +
columnConsumerGroup + `.` + columnDelaySeconds + "," +
columnConsumerGroup + `.` + columnDeadLetterQueueDestinationUUID + "," +
columnConsumerGroup + `.` + columnOwnerEmail + "," +
columnConsumerGroup + `.` + columnIsMultiZone + "," +
columnConsumerGroup + `.` + columnActiveZone + "," +
columnConsumerGroup + `.` + columnZoneConfigs + "," +
columnConsumerGroup + `.` + columnOptions +
sqlConsumerGroupType +
` FROM ` + tableConsumerGroupsByName +
` WHERE ` + columnDestinationUUID + `=? and ` + columnName + `=?`

sqlGetCG = `SELECT ` +
columnConsumerGroup + `.` + columnUUID + "," +
columnConsumerGroup + `.` + columnDestinationUUID + "," +
columnConsumerGroup + `.` + columnName + "," +
columnConsumerGroup + `.` + columnStartFrom + "," +
columnConsumerGroup + `.` + columnStatus + "," +
columnConsumerGroup + `.` + columnLockTimeoutSeconds + "," +
columnConsumerGroup + `.` + columnMaxDeliveryCount + "," +
columnConsumerGroup + `.` + columnSkipOlderMessagesSeconds + "," +
columnConsumerGroup + `.` + columnDelaySeconds + "," +
columnConsumerGroup + `.` + columnDeadLetterQueueDestinationUUID + "," +
columnConsumerGroup + `.` + columnOwnerEmail + "," +
columnConsumerGroup + `.` + columnIsMultiZone + "," +
columnConsumerGroup + `.` + columnActiveZone + "," +
columnConsumerGroup + `.` + columnZoneConfigs + "," +
columnConsumerGroup + `.` + columnOptions +
sqlConsumerGroupType +
` FROM ` + tableConsumerGroups

sqlGetCGByUUID = sqlGetCG + ` WHERE ` + columnUUID + `=?`

sqlListCGsByDestUUID = `SELECT ` +
columnConsumerGroup + `.` + columnUUID + "," +
columnDestinationUUID + "," +
columnName + "," +
columnConsumerGroup + `.` + columnStartFrom + "," +
columnConsumerGroup + `.` + columnStatus + "," +
columnConsumerGroup + `.` + columnLockTimeoutSeconds + "," +
columnConsumerGroup + `.` + columnMaxDeliveryCount + "," +
columnConsumerGroup + `.` + columnSkipOlderMessagesSeconds + "," +
columnConsumerGroup + `.` + columnDelaySeconds + "," +
columnConsumerGroup + `.` + columnDeadLetterQueueDestinationUUID + "," +
columnConsumerGroup + `.` + columnOwnerEmail + "," +
columnConsumerGroup + `.` + columnIsMultiZone + "," +
columnConsumerGroup + `.` + columnActiveZone + "," +
columnConsumerGroup + `.` + columnZoneConfigs + "," +
columnConsumerGroup + `.` + columnOptions +
sqlConsumerGroupType +
` FROM ` + tableConsumerGroupsByName +
` WHERE ` + columnDestinationUUID + `=?`

sqlListCGsUUID = `SELECT ` +
sqlConsumerGroupType +
` FROM ` + tableConsumerGroups +
` WHERE ` + columnDestinationUUID + `=?`

sqlUpdateCGByUUID = `UPDATE ` + tableConsumerGroups +
` SET ` + columnIsMultiZone + ` = ?, ` + columnConsumerGroup + `= ` + sqlCGValue +
` SET ` + columnDestinationUUID + ` = ?, ` + columnIsMultiZone + ` = ?, ` + columnConsumerGroup + `= ` + sqlCGValue +
` WHERE ` + columnUUID + `=?`

sqlUpdateCGByName = `UPDATE ` + tableConsumerGroupsByName +
Expand Down Expand Up @@ -1334,6 +1314,7 @@ func (s *CassandraMetadataService) CreateConsumerGroupUUID(ctx thrift.Context, r

err = s.session.Query(sqlInsertCGByUUID,
cgUUID,
dstUUID,
createRequest.GetIsMultiZone(),
cgUUID,
dstUUID,
Expand Down Expand Up @@ -1651,6 +1632,7 @@ func (s *CassandraMetadataService) UpdateConsumerGroup(ctx thrift.Context, reque

batch.Query(sqlUpdateCGByUUID,
// Value columns
newCG.GetDestinationUUID(),
newCG.GetIsMultiZone(),
newCG.GetConsumerGroupUUID(),
newCG.GetDestinationUUID(),
Expand Down Expand Up @@ -1794,6 +1776,7 @@ func (s *CassandraMetadataService) DeleteConsumerGroup(ctx thrift.Context, reque

batch.Query(sqlUpdateCGByUUID,
// Value columns
existingCG.GetDestinationUUID(),
existingCG.GetIsMultiZone(),
existingCG.GetConsumerGroupUUID(),
existingCG.GetDestinationUUID(),
Expand Down Expand Up @@ -1862,6 +1845,7 @@ func (s *CassandraMetadataService) DeleteConsumerGroupUUID(ctx thrift.Context, r

query := s.session.Query(sqlDeleteCGByUUIDWithTTL,
existing.GetConsumerGroupUUID(),
existing.GetDestinationUUID(),
existing.GetIsMultiZone(),
existing.GetConsumerGroupUUID(),
existing.GetDestinationUUID(),
Expand Down Expand Up @@ -2003,6 +1987,73 @@ func (s *CassandraMetadataService) ListConsumerGroups(ctx thrift.Context, reques
return result, nil
}

// ListConsumerGroupsUUID returns all ConsumerGroups matching the given destination-uuid.
func (s *CassandraMetadataService) ListConsumerGroupsUUID(ctx thrift.Context, request *shared.ListConsumerGroupsUUIDRequest) (*shared.ListConsumerGroupsUUIDResult_, error) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you add some UT for the new API?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i have now.


if !request.IsSetDestinationUUID() {
return nil, &shared.BadRequestError{
Message: fmt.Sprintf("DestinationUUID not specified"),
}
}

dstUUID := request.GetDestinationUUID()

var iter *gocql.Iter
iter = s.session.Query(sqlListCGsUUID, dstUUID).Consistency(s.lowConsLevel).PageSize(int(request.GetLimit())).PageState(request.PageToken).Iter()

if iter == nil {
return nil, &shared.InternalServiceError{
Message: "Query returned nil iterator",
}
}

result := &shared.ListConsumerGroupsUUIDResult_{
ConsumerGroups: []*shared.ConsumerGroupDescription{},
NextPageToken: request.PageToken,
}
cg := getUtilConsumerGroupDescription()
var zoneConfigsData []map[string]interface{}
for iter.Scan(
cg.ConsumerGroupUUID,
cg.DestinationUUID,
cg.ConsumerGroupName,
cg.StartFrom,
cg.Status,
cg.LockTimeoutSeconds,
cg.MaxDeliveryCount,
cg.SkipOlderMessagesSeconds,
cg.DelaySeconds,
&cg.DeadLetterQueueDestinationUUID,
cg.OwnerEmail,
cg.IsMultiZone,
cg.ActiveZone,
&zoneConfigsData,
&cg.Options) {

// skip over deleted rows
if cg.GetStatus() == shared.ConsumerGroupStatus_DELETED {
zoneConfigsData = nil
continue
}

cg.ZoneConfigs = unmarshalCgZoneConfigs(zoneConfigsData)
result.ConsumerGroups = append(result.ConsumerGroups, cg)
cg = getUtilConsumerGroupDescription()
zoneConfigsData = nil
}

nextPageToken := iter.PageState()
result.NextPageToken = make([]byte, len(nextPageToken))
copy(result.NextPageToken, nextPageToken)
if err := iter.Close(); err != nil {
return nil, &shared.InternalServiceError{
Message: err.Error(),
}
}

return result, nil
}

// ListAllConsumerGroups returns all ConsumerGroups in ConsumerGroups Table. This API is only used for debuging tool
func (s *CassandraMetadataService) ListAllConsumerGroups(ctx thrift.Context, request *shared.ListConsumerGroupRequest) (*shared.ListConsumerGroupResult_, error) {

Expand Down Expand Up @@ -3394,7 +3445,10 @@ func (s *CassandraMetadataService) UpdateStoreExtentReplicaStats(ctx thrift.Cont
)
}
if err := s.session.ExecuteBatch(batch); err != nil {
s.log.WithField(common.TagExt, request.GetExtentUUID()).Error("UpdateExtentReplicaStats failed")
s.log.WithFields(bark.Fields{
common.TagExt: request.GetExtentUUID(),
common.TagErr: err,
}).Error("UpdateExtentReplicaStats failed")
return &shared.InternalServiceError{
Message: "UpdateStoreExtentReplicaStats: %v" + err.Error(),
}
Expand Down
Loading