use most reasonable interpretation of time

common/util.go

@@ -700,3 +700,28 @@ moreOverrides:
 
 	return defaultVal
 }
+
+// FindNearestInt finds the integer that is closest to the given 'target'
+func FindNearestInt(target int64, nums ...int64) (nearest int64) {
+
+	nearest = math.MaxInt64
+	minΔ := uint64(math.MaxUint64)
+
+	for _, num := range nums {
+
+		var Δ uint64 // absolute difference
+
+		if num > target {
+			Δ = uint64(num) - uint64(target)
+		} else {
+			Δ = uint64(target) - uint64(num)
+		}
+
+		if Δ < minΔ {
+			minΔ = Δ
+			nearest = num
+		}
+	}
+
+	return
+}

services/frontendhost/frontend.go

@@ -301,31 +301,38 @@ func convertCGZoneConfigToInternal(cgZoneCfg *c.ConsumerGroupZoneConfig) *shared
 	return internalCGZoneCfg
 }
 
+// interpretTimeNanos converts the given timestamp to nanosecond units after
+// interpreting its units based on what yields the nearest time to 'now'.
+func interpretTimeNanos(ts int64) int64 {
+
+	// tsNanos, tsMicros, tsMillis and tsSeconds are what 'ts' would be
+	// would be if it were assumed to be in {nano,micro,milli,}seconds
+	// respectively and they are each converted to nanosecond units.  we
+	// use the timestamp that is closest to 'now' as the most reasonable
+	// interpretation of the given 'ts'. we can safely ignore potential
+	// overflows in our computations, because that would result in a
+	// negative number that would push it further from the 'now' timestamp.
+
+	tsNanos, tsMicros, tsMillis, tsSeconds := ts, ts*1e3, ts*1e6, ts*1e9
+
+	// 'now' in unix-nanos
+	now := time.Now().UnixNano()
+
+	// the time that is nearest to 'now' (in nanoseconds), would be the most
+	// reasonable interpretation of 'ts'
+	return common.FindNearestInt(now, tsNanos, tsMicros, tsMillis, tsSeconds)
+}
+
 // convertCreateCGRequestToInternal converts Cherami CreateConsumerGroupRequest to internal shared CreateConsumerGroupRequest
 func convertCreateCGRequestToInternal(createRequest *c.CreateConsumerGroupRequest) *shared.CreateConsumerGroupRequest {
 
-	// detect and correct the units for 'startFrom' (expected internally to be in nanoseconds)
+	// detect and correct the units for 'startFrom' (expected internally to be in nanoseconds);
+	// for '0'
 	startFrom := createRequest.GetStartFrom()
 
-	// Note: 2000-01-01T00:00 is 946684800 seconds since epoch
-	switch {
-	case startFrom == 0:
-		// special-case '0' and let it through
-
-	case startFrom < 946684800000:
-		// likely in seconds; convert to nanoseconds
-		startFrom *= 1000000000
-
-	case startFrom < 946684800000000:
-		// likely in milliseconds; convert to nanoseconds
-		startFrom *= 1000000
-
-	case startFrom < 946684800000000000:
-		// likely in microseconds; convert to nanoseconds
-		startFrom *= 1000
-
-	default:
-		// likely in nanoseconds already
+	// special-case a StartFrom of '0' (start from beginning)
+	if startFrom != 0 {
+		startFrom = interpretTimeNanos(startFrom)
 	}
 
 	internalCreateRequest := shared.NewCreateConsumerGroupRequest()

services/frontendhost/frontend_test.go

@@ -671,24 +671,12 @@ func (s *FrontendHostSuite) TestFrontendHostCreateConsumerGroupStartFrom() {
 	testCG := s.generateKey("/bar/CGName")
 	frontendHost, ctx := s.utilGetContextAndFrontend()
 
-	// set start-from to, say, four weeks ago
-	startFromTime := time.Now().Add(-4 * 7 * 24 * time.Hour)
+	testStartFrom := func(startFrom, startFromExpected int64) {
 
-	startFromSeconds := startFromTime.Unix()
-	startFromMillis := startFromSeconds * int64(1000)
-	startFromMicros := startFromSeconds * int64(1000000)
-	startFromNanos := startFromSeconds * int64(1000000000)
-
-	testStartFrom := func(startFrom int64) {
 		req := c.NewCreateConsumerGroupRequest()
 		req.DestinationPath = common.StringPtr(testPath)
 		req.ConsumerGroupName = common.StringPtr(testCG)
-
-		if startFrom == 0 {
-			req.StartFrom = common.Int64Ptr(0)
-		} else {
-			req.StartFrom = common.Int64Ptr(startFromNanos)
-		}
+		req.StartFrom = common.Int64Ptr(startFromExpected)
 
 		cgDesc := cgCreateRequestToDesc(req)
 		frontendHost.writeCacheDestinationPathForUUID(destinationUUID(cgDesc.GetDestinationUUID()), testPath)
@@ -705,14 +693,9 @@ func (s *FrontendHostSuite) TestFrontendHostCreateConsumerGroupStartFrom() {
 		})
 
 		s.mockController.On("CreateConsumerGroup", mock.Anything, mock.Anything).Once().Return(cgDesc, nil).Run(func(args mock.Arguments) {
-			createReq := args.Get(1).(*shared.CreateConsumerGroupRequest)
-
-			// we should expect to see the startFrom in nano-seconds in every case
-			if startFrom == 0 {
-				s.EqualValues(0, createReq.GetStartFrom())
-			} else {
-				s.EqualValues(startFromNanos, createReq.GetStartFrom())
-			}
+
+			// ensure we get the value expected
+			s.EqualValues(startFromExpected, args.Get(1).(*shared.CreateConsumerGroupRequest).GetStartFrom())
 		})
 
 		// set startFrom to the test value (of varying units)
@@ -721,12 +704,25 @@ func (s *FrontendHostSuite) TestFrontendHostCreateConsumerGroupStartFrom() {
 		frontendHost.CreateConsumerGroup(ctx, req)
 	}
 
-	// test startFrom with varying units
-	testStartFrom(startFromNanos)
-	testStartFrom(startFromMicros)
-	testStartFrom(startFromMillis)
-	testStartFrom(startFromSeconds)
-	testStartFrom(0)
+	testStartFromTime := func(startFromTime time.Time) {
+
+		startFromSeconds := startFromTime.Unix()
+		startFromMillis := startFromSeconds * int64(1e3)
+		startFromMicros := startFromSeconds * int64(1e6)
+		startFromNanos := startFromSeconds * int64(1e9)
+
+		// test startFrom with varying units
+		testStartFrom(startFromSeconds, startFromNanos)
+		testStartFrom(startFromMillis, startFromNanos)
+		testStartFrom(startFromMicros, startFromNanos)
+		testStartFrom(startFromNanos, startFromNanos)
+	}
+
+	testStartFromTime(time.Now().Add(-4 * 7 * 24 * time.Hour))                 // four weeks ago
+	testStartFromTime(time.Date(2016, time.January, 1, 0, 0, 0, 0, time.UTC))  // ~ a year ago
+	testStartFromTime(time.Date(2046, time.June, 30, 11, 59, 59, 0, time.UTC)) // ~30 years from now
+
+	testStartFrom(0, 0) // start-from of '0'
 }
 
 // TestFrontendHostReadConsumerGroupRejectNil tests that a nil request fails with BadRequestError