Fix export bug and more updates to checkpointing.

Docker/kinesis_entrypoint.sh

@@ -20,10 +20,6 @@ fi
 if [ -n "$CHECKPOINT_TABLE" ]
 then
     BOOL_FLAGS=${BOOL_FLAGS}" --checkpoint-table=${CHECKPOINT_TABLE}"
-	if [ -n "$POST_CHECKPOINT_INIT_DELAY" ]
-	then
-    	BOOL_FLAGS=${BOOL_FLAGS}" --post-checkpoint-init-delay=${POST_CHECKPOINT_INIT_DELAY}"
-	fi
 fi
 if [ -n "$LOG_LEVEL" ]
 then

quanta-kinesis-consumer-lib/q-kinesis-lib.go

@@ -61,7 +61,6 @@ type Main struct {
 	AssumeRoleArn       string
 	AssumeRoleArnRegion string
 	Deaggregate         bool
-	PostCheckpointInitDelay int
 	Collate             bool
 	ShardKey            string
 	ProtoConfig			string
@@ -180,32 +179,19 @@ func (m *Main) Init(customEndpoint string) (int, error) {
 		return 0, err
 	}
 	shardCount := len(shout.Shards)
-	initializedShardsInDB := false
+	foundCheckpointRecords := false
 	if db != nil && m.InitialPos != "TRIM_HORIZON"  {
 		for _, v := range shout.Shards {
 			seq, _ := db.GetCheckpoint(*streamName, *v.ShardId)
 			if seq != "" && seq != "0" {
+				foundCheckpointRecords = true
 				continue
 			}
-			sequenceRange := *v.SequenceNumberRange
-			sequenceNumber := *sequenceRange.StartingSequenceNumber
-			if sequenceRange.EndingSequenceNumber != nil {
-				sequenceNumber = *sequenceRange.EndingSequenceNumber
-			}
-			if sequenceNumber == "" {
-				sequenceNumber = *sequenceRange.StartingSequenceNumber
-			}
-			err := db.SetCheckpoint(*streamName, *v.ShardId, sequenceNumber)
-			//u.Debugf("Initializing checkpoint for shard %s.%s, SEQ = %s", *streamName, *v.ShardId,
-			//	 sequenceNumber)
-        	if err != nil {
-				return 0, fmt.Errorf("failed to set inital checkpoint, %v", err)
-        	}
-			initializedShardsInDB = true
 		}
 	}
-	if initializedShardsInDB {
-		time.Sleep(time.Duration(m.PostCheckpointInitDelay) * time.Second)
+	if m.InitialPos == "LATEST" && foundCheckpointRecords {
+		u.Errorf("Checkpoint enabled and records exist.  Shard iterator is 'LATEST' setting it to 'AFTER_SEQUENCE_NUMBER'.")
+		m.InitialPos = "AFTER_SEQUENCE_NUMBER"
 	}
 
 	m.metrics = cloudwatch.New(sess)
@@ -360,9 +346,8 @@ func (m *Main) MainProcessingLoop() error {
 			u.Warnf("Received Cancellation.")
 		}
 		if m.InitialPos == "TRIM_HORIZON" {
-			u.Error("can't re-initialize 'in-place' if set to TRIM_HORIZON, exiting")
-			// os.Exit(1)
-			return fmt.Errorf("can't re-initialize 'in-place' if set to TRIM_HORIZON")
+			u.Error("initial position was TRIM_HORIZON, re-initializing with set to LATEST ")
+			m.InitialPos = "LATEST"
 		}
 		u.Warnf("Re-initializing.")
 		var err error

quanta-kinesis-consumer/quanta-kinesis-consumer.go

@@ -49,7 +49,6 @@ func main() {
 	deaggregate := app.Flag("deaggregate", "Incoming payload records are aggregated.").Bool()
 	scanInterval := app.Flag("scan-interval", "Scan interval (milliseconds)").Default("1000").Int()
 	protoPath := app.Flag("proto-path", "Path to protobuf descriptor files root directory.").String()
-	postCheckpointInitDelay := app.Flag("post-checkpoint-init-delay", "Delay seconds after checkpoint table init.").Default("30").Int()
 	logLevel := app.Flag("log-level", "Log Level [ERROR, WARN, INFO, DEBUG]").Default("WARN").String()
 
 	kingpin.MustParse(app.Parse(os.Args[1:]))
@@ -66,7 +65,6 @@ func main() {
 	main.ScanInterval = *scanInterval
 	main.Port = int(*port)
 	main.ConsulAddr = *consul
-	main.PostCheckpointInitDelay = *postCheckpointInitDelay
 
 	log.Printf("Set Logging level to %v.", *logLevel)
 	log.Printf("Kinesis stream %v.", main.Stream)
@@ -95,10 +93,6 @@ func main() {
 			os.Exit(1)
 		}
 		log.Printf("DynamoDB checkpoint table name [%s]", main.CheckpointTable)
-		if main.InitialPos == "LATEST" {
-			u.Errorf("Checkpoint enabled.  Shard iterator is 'LATEST' setting it to 'AFTER_SEQUENCE_NUMBER'.")
-			main.InitialPos = "AFTER_SEQUENCE_NUMBER"
-		}
 	}
 
 	if shardKey != nil {

sink/s3sink.go

@@ -268,8 +268,12 @@ func (s *S3ParquetSink) Open(ctx *plan.Context, bucketpath string, params map[st
 	// Create S3 service client
 	u.Infof("Parquet Sink: Opening Output S3 path s3://%s/%s", bucket, file)
 	s.outFile, err = pgs3.NewS3FileWriterWithClient(context.Background(), s.s3svc, bucket, file, nil, func(p *s3.PutObjectInput) {
-		p.SSEKMSKeyId = aws.String(s.sseKmsKeyID)
-		p.ServerSideEncryption = "aws:kms"
+		if s.sseKmsKeyID != "" {
+			p.SSEKMSKeyId = aws.String(s.sseKmsKeyID)
+			p.ServerSideEncryption = "aws:kms"
+		} else {
+			p.SSEKMSKeyId = nil
+		}
 		p.ACL = types.ObjectCannedACL(s.acl)
 	})
 
@@ -288,8 +292,10 @@ func (s *S3ParquetSink) Open(ctx *plan.Context, bucketpath string, params map[st
 			s.md[i] = fmt.Sprintf("name=%s, type=FLOAT", v.As)
 		case value.BoolType:
 			s.md[i] = fmt.Sprintf("name=%s, type=BOOLEAN", v.As)
+		case value.TimeType:
+			s.md[i] = fmt.Sprintf("name=%s, type=INT64, convertedtype=TIMESTAMP_MICROS", v.As)
 		default:
-			s.md[i] = fmt.Sprintf("name=%s, type=UTF8, encoding=PLAIN_DICTIONARY", v.As)
+			s.md[i] = fmt.Sprintf("name=%s, type=BYTE_ARRAY, convertedtype=utf8, encoding=PLAIN_DICTIONARY", v.As)
 		}
 	}
 

version.txt

@@ -1,3 +1,3 @@
-quanta 0.9.15-rc-6
+quanta 0.9.15-rc-7