Delay-queue: changes to thrift, metadata, cli, etc

clients/metadata/metadata_cassandra.go

@@ -101,6 +101,7 @@ const (
 	columnConsumerGroupUUID              = "consumer_group_uuid"
 	columnConsumerGroupVisibility        = "consumer_group_visibility"
 	columnCreatedTime                    = "created_time"
+	columnDelaySeconds                   = "delay_seconds"
 	columnDLQConsumerGroup               = "dlq_consumer_group"
 	columnDLQMergeBefore                 = "dlq_merge_before"
 	columnDLQPurgeBefore                 = "dlq_purge_before"
@@ -544,6 +545,7 @@ func getUtilConsumerGroupDescription() *shared.ConsumerGroupDescription {
 	result.LockTimeoutSeconds = common.Int32Ptr(0)
 	result.MaxDeliveryCount = common.Int32Ptr(0)
 	result.SkipOlderMessagesSeconds = common.Int32Ptr(0)
+	result.DelaySeconds = common.Int32Ptr(0)
 	result.OwnerEmail = common.StringPtr("")
 	result.IsMultiZone = common.BoolPtr(false)
 	result.ActiveZone = common.StringPtr("")
@@ -1137,6 +1139,7 @@ const (
 		columnLockTimeoutSeconds + `: ?, ` +
 		columnMaxDeliveryCount + `: ?, ` +
 		columnSkipOlderMessagesSeconds + `: ?, ` +
+		columnDelaySeconds + `: ?, ` +
 		columnDeadLetterQueueDestinationUUID + `: ?, ` +
 		columnOwnerEmail + `: ?, ` +
 		columnIsMultiZone + `: ?, ` +
@@ -1167,6 +1170,7 @@ const (
 		columnConsumerGroup + `.` + columnLockTimeoutSeconds + "," +
 		columnConsumerGroup + `.` + columnMaxDeliveryCount + "," +
 		columnConsumerGroup + `.` + columnSkipOlderMessagesSeconds + "," +
+		columnConsumerGroup + `.` + columnDelaySeconds + "," +
 		columnConsumerGroup + `.` + columnDeadLetterQueueDestinationUUID + "," +
 		columnConsumerGroup + `.` + columnOwnerEmail + "," +
 		columnConsumerGroup + `.` + columnIsMultiZone + "," +
@@ -1184,6 +1188,7 @@ const (
 		columnConsumerGroup + `.` + columnLockTimeoutSeconds + "," +
 		columnConsumerGroup + `.` + columnMaxDeliveryCount + "," +
 		columnConsumerGroup + `.` + columnSkipOlderMessagesSeconds + "," +
+		columnConsumerGroup + `.` + columnDelaySeconds + "," +
 		columnConsumerGroup + `.` + columnDeadLetterQueueDestinationUUID + "," +
 		columnConsumerGroup + `.` + columnOwnerEmail + "," +
 		columnConsumerGroup + `.` + columnIsMultiZone + "," +
@@ -1202,6 +1207,7 @@ const (
 		columnConsumerGroup + `.` + columnLockTimeoutSeconds + "," +
 		columnConsumerGroup + `.` + columnMaxDeliveryCount + "," +
 		columnConsumerGroup + `.` + columnSkipOlderMessagesSeconds + "," +
+		columnConsumerGroup + `.` + columnDelaySeconds + "," +
 		columnConsumerGroup + `.` + columnDeadLetterQueueDestinationUUID + "," +
 		columnConsumerGroup + `.` + columnOwnerEmail + "," +
 		columnConsumerGroup + `.` + columnIsMultiZone + "," +
@@ -1291,6 +1297,7 @@ func (s *CassandraMetadataService) CreateConsumerGroupUUID(ctx thrift.Context, r
 		createRequest.GetLockTimeoutSeconds(),
 		createRequest.GetMaxDeliveryCount(),
 		createRequest.GetSkipOlderMessagesSeconds(),
+		createRequest.GetDelaySeconds(),
 		dlqUUID,
 		createRequest.GetOwnerEmail(),
 		createRequest.GetIsMultiZone(),
@@ -1316,6 +1323,7 @@ func (s *CassandraMetadataService) CreateConsumerGroupUUID(ctx thrift.Context, r
 		createRequest.GetLockTimeoutSeconds(),
 		createRequest.GetMaxDeliveryCount(),
 		createRequest.GetSkipOlderMessagesSeconds(),
+		createRequest.GetDelaySeconds(),
 		dlqUUID,
 		createRequest.GetOwnerEmail(),
 		createRequest.GetIsMultiZone(),
@@ -1359,6 +1367,7 @@ func (s *CassandraMetadataService) CreateConsumerGroupUUID(ctx thrift.Context, r
 		LockTimeoutSeconds:             common.Int32Ptr(createRequest.GetLockTimeoutSeconds()),
 		MaxDeliveryCount:               common.Int32Ptr(createRequest.GetMaxDeliveryCount()),
 		SkipOlderMessagesSeconds:       common.Int32Ptr(createRequest.GetSkipOlderMessagesSeconds()),
+		DelaySeconds:                   common.Int32Ptr(createRequest.GetDelaySeconds()),
 		DeadLetterQueueDestinationUUID: dlqUUID,
 		OwnerEmail:                     common.StringPtr(createRequest.GetOwnerEmail()),
 		IsMultiZone:                    common.BoolPtr(createRequest.GetIsMultiZone()),
@@ -1415,6 +1424,7 @@ func (s *CassandraMetadataService) readConsumerGroupByDstUUID(dstUUID string, cg
 		result.LockTimeoutSeconds,
 		result.MaxDeliveryCount,
 		result.SkipOlderMessagesSeconds,
+		result.DelaySeconds,
 		&result.DeadLetterQueueDestinationUUID,
 		result.OwnerEmail,
 		result.IsMultiZone,
@@ -1485,6 +1495,7 @@ func (s *CassandraMetadataService) ReadConsumerGroupByUUID(ctx thrift.Context, r
 		result.LockTimeoutSeconds,
 		result.MaxDeliveryCount,
 		result.SkipOlderMessagesSeconds,
+		result.DelaySeconds,
 		&result.DeadLetterQueueDestinationUUID,
 		result.OwnerEmail,
 		result.IsMultiZone,
@@ -1524,6 +1535,11 @@ func updateCGDescIfChanged(req *shared.UpdateConsumerGroupRequest, cgDesc *share
 		cgDesc.SkipOlderMessagesSeconds = common.Int32Ptr(req.GetSkipOlderMessagesSeconds())
 	}
 
+	if req.IsSetDelaySeconds() && req.GetDelaySeconds() != cgDesc.GetDelaySeconds() {
+		isChanged = true
+		cgDesc.DelaySeconds = common.Int32Ptr(req.GetDelaySeconds())
+	}
+
 	if req.IsSetStatus() && req.GetStatus() != cgDesc.GetStatus() {
 		isChanged = true
 		cgDesc.Status = common.InternalConsumerGroupStatusPtr(req.GetStatus())
@@ -1587,6 +1603,7 @@ func (s *CassandraMetadataService) UpdateConsumerGroup(ctx thrift.Context, reque
 		newCG.GetLockTimeoutSeconds(),
 		newCG.GetMaxDeliveryCount(),
 		newCG.GetSkipOlderMessagesSeconds(),
+		newCG.GetDelaySeconds(),
 		newCG.DeadLetterQueueDestinationUUID, // May be null
 		newCG.GetOwnerEmail(),
 		newCG.GetIsMultiZone(),
@@ -1606,6 +1623,7 @@ func (s *CassandraMetadataService) UpdateConsumerGroup(ctx thrift.Context, reque
 		newCG.GetLockTimeoutSeconds(),
 		newCG.GetMaxDeliveryCount(),
 		newCG.GetSkipOlderMessagesSeconds(),
+		newCG.GetDelaySeconds(),
 		newCG.DeadLetterQueueDestinationUUID, // May be null
 		newCG.GetOwnerEmail(),
 		newCG.GetIsMultiZone(),
@@ -1725,6 +1743,7 @@ func (s *CassandraMetadataService) DeleteConsumerGroup(ctx thrift.Context, reque
 		existingCG.GetLockTimeoutSeconds(),
 		existingCG.GetMaxDeliveryCount(),
 		existingCG.GetSkipOlderMessagesSeconds(),
+		existingCG.GetDelaySeconds(),
 		existingCG.DeadLetterQueueDestinationUUID, // May be Null
 		existingCG.GetOwnerEmail(),
 		existingCG.GetIsMultiZone(),
@@ -1828,6 +1847,7 @@ func (s *CassandraMetadataService) ListConsumerGroups(ctx thrift.Context, reques
 		cg.LockTimeoutSeconds,
 		cg.MaxDeliveryCount,
 		cg.SkipOlderMessagesSeconds,
+		cg.DelaySeconds,
 		&cg.DeadLetterQueueDestinationUUID,
 		cg.OwnerEmail,
 		cg.IsMultiZone,
@@ -1890,6 +1910,7 @@ func (s *CassandraMetadataService) ListAllConsumerGroups(ctx thrift.Context, req
 		cg.LockTimeoutSeconds,
 		cg.MaxDeliveryCount,
 		cg.SkipOlderMessagesSeconds,
+		cg.DelaySeconds,
 		&cg.DeadLetterQueueDestinationUUID,
 		cg.OwnerEmail,
 		cg.IsMultiZone,

clients/metadata/metadata_cassandra_test.go

@@ -1027,6 +1027,7 @@ func (s *CassandraSuite) TestMoveExtent() {
 		LockTimeoutSeconds:       common.Int32Ptr(10),
 		MaxDeliveryCount:         common.Int32Ptr(5),
 		SkipOlderMessagesSeconds: common.Int32Ptr(60),
+		DelaySeconds:             common.Int32Ptr(60),
 		OwnerEmail:               common.StringPtr("[email protected]"),
 	}
 
@@ -1671,6 +1672,7 @@ func assertConsumerGroupsEqual(s *CassandraSuite, expected, got *shared.Consumer
 	s.Equal(expected.GetLockTimeoutSeconds(), got.GetLockTimeoutSeconds(), "Wrong LockTimeoutSeconds")
 	s.Equal(expected.GetMaxDeliveryCount(), got.GetMaxDeliveryCount(), "Wrong MaxDeliveryCount")
 	s.Equal(expected.GetSkipOlderMessagesSeconds(), got.GetSkipOlderMessagesSeconds(), "Wrong SkipOlderMessagesSeconds")
+	s.Equal(expected.GetDelaySeconds(), got.GetDelaySeconds(), "Wrong DelaySeconds")
 	s.Equal(expected.GetOwnerEmail(), got.GetOwnerEmail(), "Wrong OwnerEmail")
 	s.Equal(expected.GetDeadLetterQueueDestinationUUID(), got.GetDeadLetterQueueDestinationUUID(), "Wrong DeadLetterQueueDestinationUUID")
 	s.Equal(expected.GetActiveZone(), got.GetActiveZone(), "Wrong ActiveZone")
@@ -1693,6 +1695,7 @@ func (s *CassandraSuite) TestDeleteConsumerGroupDeletesDLQ() {
 		LockTimeoutSeconds:       common.Int32Ptr(10),
 		MaxDeliveryCount:         common.Int32Ptr(5),
 		SkipOlderMessagesSeconds: common.Int32Ptr(60),
+		DelaySeconds:             common.Int32Ptr(60),
 		OwnerEmail:               common.StringPtr("[email protected]"),
 	}
 
@@ -1759,6 +1762,7 @@ func (s *CassandraSuite) TestConsumerGroupCRUD() {
 		LockTimeoutSeconds:       common.Int32Ptr(10),
 		MaxDeliveryCount:         common.Int32Ptr(5),
 		SkipOlderMessagesSeconds: common.Int32Ptr(60),
+		DelaySeconds:             common.Int32Ptr(60),
 		OwnerEmail:               common.StringPtr("[email protected]"),
 		IsMultiZone:              common.BoolPtr(true),
 		ActiveZone:               common.StringPtr("zone1"),
@@ -1773,6 +1777,7 @@ func (s *CassandraSuite) TestConsumerGroupCRUD() {
 		LockTimeoutSeconds:       common.Int32Ptr(createReq.GetLockTimeoutSeconds()),
 		MaxDeliveryCount:         common.Int32Ptr(createReq.GetMaxDeliveryCount()),
 		SkipOlderMessagesSeconds: common.Int32Ptr(createReq.GetSkipOlderMessagesSeconds()),
+		DelaySeconds:             common.Int32Ptr(createReq.GetDelaySeconds()),
 		OwnerEmail:               common.StringPtr(createReq.GetOwnerEmail()),
 		IsMultiZone:              common.BoolPtr(createReq.GetIsMultiZone()),
 		ActiveZone:               common.StringPtr(createReq.GetActiveZone()),
@@ -1833,6 +1838,7 @@ func (s *CassandraSuite) TestConsumerGroupCRUD() {
 			LockTimeoutSeconds:       common.Int32Ptr(99),
 			MaxDeliveryCount:         common.Int32Ptr(99),
 			SkipOlderMessagesSeconds: common.Int32Ptr(100),
+			DelaySeconds:             common.Int32Ptr(100),
 			OwnerEmail:               common.StringPtr("[email protected]"),
 		}
 
@@ -1846,6 +1852,7 @@ func (s *CassandraSuite) TestConsumerGroupCRUD() {
 		expectedCG.LockTimeoutSeconds = common.Int32Ptr(updateReq.GetLockTimeoutSeconds())
 		expectedCG.MaxDeliveryCount = common.Int32Ptr(updateReq.GetMaxDeliveryCount())
 		expectedCG.SkipOlderMessagesSeconds = common.Int32Ptr(updateReq.GetSkipOlderMessagesSeconds())
+		expectedCG.DelaySeconds = common.Int32Ptr(updateReq.GetDelaySeconds())
 		expectedCG.OwnerEmail = common.StringPtr(updateReq.GetOwnerEmail())
 		expectedCG.ActiveZone = common.StringPtr(updateReq.GetActiveZone())
 
@@ -1914,6 +1921,7 @@ func (s *CassandraSuite) TestCGCRUDOnPhantomDestination() {
 		LockTimeoutSeconds:       common.Int32Ptr(10),
 		MaxDeliveryCount:         common.Int32Ptr(5),
 		SkipOlderMessagesSeconds: common.Int32Ptr(6),
+		DelaySeconds:             common.Int32Ptr(6),
 		OwnerEmail:               common.StringPtr("[email protected]"),
 	}
 
@@ -1929,6 +1937,7 @@ func (s *CassandraSuite) TestCGCRUDOnPhantomDestination() {
 		LockTimeoutSeconds:       common.Int32Ptr(99),
 		MaxDeliveryCount:         common.Int32Ptr(99),
 		SkipOlderMessagesSeconds: common.Int32Ptr(99),
+		DelaySeconds:             common.Int32Ptr(99),
 		OwnerEmail:               common.StringPtr("[email protected]"),
 	}
 	_, err = s.client.UpdateConsumerGroup(nil, updateReq)
@@ -1968,6 +1977,7 @@ func (s *CassandraSuite) TestReCreateConsumerGroup() {
 		LockTimeoutSeconds:       common.Int32Ptr(10),
 		MaxDeliveryCount:         common.Int32Ptr(5),
 		SkipOlderMessagesSeconds: common.Int32Ptr(60),
+		DelaySeconds:             common.Int32Ptr(60),
 		OwnerEmail:               common.StringPtr("[email protected]"),
 	}
 
@@ -2019,6 +2029,7 @@ func (s *CassandraSuite) TestListConsumerGroups() {
 			LockTimeoutSeconds:       common.Int32Ptr(10),
 			MaxDeliveryCount:         common.Int32Ptr(5),
 			SkipOlderMessagesSeconds: common.Int32Ptr(60),
+			DelaySeconds:             common.Int32Ptr(60),
 			OwnerEmail:               common.StringPtr("[email protected]"),
 		}
 
@@ -2099,6 +2110,7 @@ func (s *CassandraSuite) TestListAllConsumerGroups() {
 			LockTimeoutSeconds:       common.Int32Ptr(10),
 			MaxDeliveryCount:         common.Int32Ptr(5),
 			SkipOlderMessagesSeconds: common.Int32Ptr(60),
+			DelaySeconds:             common.Int32Ptr(60),
 			OwnerEmail:               common.StringPtr("[email protected]"),
 		}
 
@@ -2128,6 +2140,7 @@ func (s *CassandraSuite) TestListAllConsumerGroups() {
 					s.Equal(int32(10), gotCG.GetLockTimeoutSeconds())
 					s.Equal(int32(5), gotCG.GetMaxDeliveryCount())
 					s.Equal(int32(60), gotCG.GetSkipOlderMessagesSeconds())
+					s.Equal(int32(60), gotCG.GetDelaySeconds())
 					s.Equal(string("[email protected]"), gotCG.GetOwnerEmail())
 					delete(groupMap, gotCG.GetConsumerGroupName())
 				}
@@ -2164,6 +2177,7 @@ func (s *CassandraSuite) TestSameCGNameOnDifferentDestinations() {
 		LockTimeoutSeconds:       common.Int32Ptr(10),
 		MaxDeliveryCount:         common.Int32Ptr(5),
 		SkipOlderMessagesSeconds: common.Int32Ptr(60),
+		DelaySeconds:             common.Int32Ptr(60),
 		OwnerEmail:               common.StringPtr("[email protected]"),
 	}
 

clients/metadata/schema/metadata.cql

@@ -214,6 +214,7 @@ CREATE TYPE consumer_group (
   lock_timeout_seconds int,
   max_delivery_count int,
   skip_older_messages_seconds int,
+  delay_seconds int,
   dead_letter_queue_destination_uuid uuid,
   owner_email text,
   is_multi_zone boolean,

clients/metadata/schema/v15/201706160000_delay.cql

@@ -0,0 +1 @@
+ALTER TYPE consumer_group ADD delay_seconds int;

clients/metadata/schema/v15/manifest.json

@@ -0,0 +1,8 @@
+{
+	"CurrVersion": 15,
+	"MinCompatibleVersion": 8,
+	"Description": "Add Delay",
+	"SchemaUpdateCqlFiles": [
+		"201706160000_delay.cql"
+	]
+}

cmd/tools/admin/main.go

@@ -140,6 +140,11 @@ func main() {
 							Value: 7200,
 							Usage: "Skip messages older than this duration in seconds.",
 						},
+						cli.IntFlag{
+							Name:  "delay_seonds, d",
+							Value: 0,
+							Usage: "Delay to add to every message, in seconds.",
+						},
 						cli.StringFlag{
 							Name:  "owner_email, oe",
 							Value: "",
@@ -343,6 +348,11 @@ func main() {
 							Value: 7200,
 							Usage: "Skip messages older than this duration in seconds.",
 						},
+						cli.IntFlag{
+							Name:  "delay_seonds, d",
+							Value: 0,
+							Usage: "Delay to add to every message, in seconds.",
+						},
 						cli.StringFlag{
 							Name:  "owner_email, oe",
 							Value: "",
@@ -721,7 +731,7 @@ func main() {
 		{
 			Name:    "seal-check",
 			Aliases: []string{"sc"},
-			Usage:   "seal-check <dest> [--seal]",
+			Usage:   "seal-check <dest> [-seal]",
 			Flags: []cli.Flag{
 				cli.StringFlag{
 					Name:  "prefix, pf",
@@ -772,7 +782,7 @@ func main() {
 		{
 			Name:    "store-purgeextent",
 			Aliases: []string{"purge"},
-			Usage:   "purge <store_uuid> <extent_uuid> [<address> | --entirely]",
+			Usage:   "purge <store_uuid> <extent_uuid> [<address> | -entirely]",
 			Flags: []cli.Flag{
 				cli.BoolFlag{
 					Name:  "entirely",

cmd/tools/cli/main.go

@@ -34,8 +34,11 @@ const (
 
 	strMaxDeliveryCount = "Maximum number of times a message is delivered\n\tbefore it is sent to the dead-letter queue (DLQ)"
 
-	strSkipOlderMessagesInSeconds = `Skip messages older than this duration in seconds. (NOT IMPLEMENTED)`
-	intSkipOlderMessagesInSeconds = 999999999 // More than 30 years; placeholder for eventual implementation
+	strSkipOlderMessagesInSeconds = `Skip messages older than this duration in seconds.`
+	intSkipOlderMessagesInSeconds = 999999999 // More than 30 years
+
+	strDelaySeconds = `Delay to introduce to all messages, in seconds.`
+	intDelaySeconds = 0 // zero delay, by default
 )
 
 func main() {
@@ -128,7 +131,7 @@ func main() {
 						},
 						cli.StringSliceFlag{
 							Name:  "kafka_topics, kt",
-							Usage: "List of kafka topics to subscribe to. Use multiple times, e.g. \"--kafka_topics topic_a --kafka_topics topic_b\"",
+							Usage: "List of kafka topics to subscribe to. Use multiple times, e.g. \"-kafka_topics topic_a -kafka_topics topic_b\"",
 						},
 					},
 					Action: func(c *cli.Context) {
@@ -160,6 +163,11 @@ func main() {
 							Value: intSkipOlderMessagesInSeconds,
 							Usage: strSkipOlderMessagesInSeconds,
 						},
+						cli.IntFlag{
+							Name:  "delay_seconds, d",
+							Value: intDelaySeconds,
+							Usage: strDelaySeconds,
+						},
 						cli.StringFlag{
 							Name:  "owner_email, oe",
 							Value: cliHelper.GetDefaultOwnerEmail(),
@@ -299,6 +307,11 @@ func main() {
 							Value: intSkipOlderMessagesInSeconds,
 							Usage: strSkipOlderMessagesInSeconds,
 						},
+						cli.IntFlag{
+							Name:  "delay_seconds, d",
+							Value: intDelaySeconds,
+							Usage: strDelaySeconds,
+						},
 						cli.StringFlag{
 							Name:  "owner_email, oe",
 							Value: cliHelper.GetDefaultOwnerEmail(),

services/frontendhost/frontend.go

@@ -380,6 +380,7 @@ func convertCreateCGRequestToInternal(createRequest *c.CreateConsumerGroupReques
 	internalCreateRequest.LockTimeoutSeconds = common.Int32Ptr(createRequest.GetLockTimeoutInSeconds())
 	internalCreateRequest.MaxDeliveryCount = common.Int32Ptr(createRequest.GetMaxDeliveryCount())
 	internalCreateRequest.SkipOlderMessagesSeconds = common.Int32Ptr(createRequest.GetSkipOlderMessagesInSeconds())
+	internalCreateRequest.DelaySeconds = common.Int32Ptr(createRequest.GetDelaySeconds())
 	internalCreateRequest.StartFrom = common.Int64Ptr(startFrom)
 	internalCreateRequest.OwnerEmail = common.StringPtr(createRequest.GetOwnerEmail())
 	internalCreateRequest.IsMultiZone = common.BoolPtr(createRequest.GetIsMultiZone())
@@ -408,6 +409,9 @@ func convertUpdateCGRequestToInternal(updateRequest *c.UpdateConsumerGroupReques
 	if updateRequest.IsSetSkipOlderMessagesInSeconds() {
 		internalUpdateRequest.SkipOlderMessagesSeconds = common.Int32Ptr(updateRequest.GetSkipOlderMessagesInSeconds())
 	}
+	if updateRequest.IsSetDelaySeconds() {
+		internalUpdateRequest.DelaySeconds = common.Int32Ptr(updateRequest.GetDelaySeconds())
+	}
 	if updateRequest.IsSetStatus() {
 		internalUpdateRequest.Status = common.InternalConsumerGroupStatusPtr(shared.ConsumerGroupStatus(updateRequest.GetStatus()))
 	}
@@ -472,6 +476,7 @@ func (h *Frontend) convertConsumerGroupFromInternal(ctx thrift.Context, _cgDesc
 	cgDesc.Status = c.ConsumerGroupStatusPtr(c.ConsumerGroupStatus(_cgDesc.GetStatus()))
 	cgDesc.StartFrom = common.Int64Ptr(_cgDesc.GetStartFrom())
 	cgDesc.SkipOlderMessagesInSeconds = common.Int32Ptr(_cgDesc.GetSkipOlderMessagesSeconds())
+	cgDesc.DelaySeconds = common.Int32Ptr(_cgDesc.GetDelaySeconds())
 	cgDesc.MaxDeliveryCount = common.Int32Ptr(_cgDesc.GetMaxDeliveryCount())
 	cgDesc.LockTimeoutInSeconds = common.Int32Ptr(_cgDesc.GetLockTimeoutSeconds())
 	cgDesc.DeadLetterQueueDestinationUUID = common.StringPtr(_cgDesc.GetDeadLetterQueueDestinationUUID())