luciano/vmware-go-kcl-v2
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

improve gofmt

Browse source
This commit is contained in:
Fabiano Graças 2021-11-11 23:11:30 +01:00 committed by Tao Jiang
parent 6f0fbfe4c7
commit f9ced84cbd
7 changed files with 56 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
clientlibrary/checkpoint/dynamodb-checkpointer.go
View file

@ -61,7 +61,7 @@ type DynamoCheckpoint struct {
LeaseDuration int LeaseDuration int
svc *dynamodb.Client svc *dynamodb.Client
kclConfig *config.KinesisClientLibConfiguration kclConfig *config.KinesisClientLibConfiguration
Retries int Retries int
lastLeaseSync time.Time lastLeaseSync time.Time
} }

87
clientlibrary/config/kcl-config.go
View file

@ -70,31 +70,31 @@ func NewKinesisClientLibConfigWithCredentials(applicationName, streamName, regio
} }
// populate the KCL configuration with default values // populate the KCL configuration with default values
return &KinesisClientLibConfiguration { return &KinesisClientLibConfiguration{
ApplicationName: applicationName, ApplicationName: applicationName,
KinesisCredentials: kinesisCreds, KinesisCredentials: kinesisCreds,
DynamoDBCredentials: dynamodbCreds, DynamoDBCredentials: dynamodbCreds,
TableName: applicationName, TableName: applicationName,
EnhancedFanOutConsumerName: applicationName, EnhancedFanOutConsumerName: applicationName,
StreamName: streamName, StreamName: streamName,
RegionName: regionName, RegionName: regionName,
WorkerID: workerID, WorkerID: workerID,
InitialPositionInStream: DefaultInitialPositionInStream, InitialPositionInStream: DefaultInitialPositionInStream,
InitialPositionInStreamExtended: *newInitialPosition(DefaultInitialPositionInStream), InitialPositionInStreamExtended: *newInitialPosition(DefaultInitialPositionInStream),
FailoverTimeMillis: DefaultFailoverTimeMillis, FailoverTimeMillis: DefaultFailoverTimeMillis,
LeaseRefreshPeriodMillis: DefaultLeaseRefreshPeriodMillis, LeaseRefreshPeriodMillis: DefaultLeaseRefreshPeriodMillis,
MaxRecords: DefaultMaxRecords, MaxRecords: DefaultMaxRecords,
IdleTimeBetweenReadsInMillis: DefaultIdleTimeBetweenReadsMillis, IdleTimeBetweenReadsInMillis: DefaultIdleTimeBetweenReadsMillis,
CallProcessRecordsEvenForEmptyRecordList: DefaultDontCallProcessRecordsForEmptyRecordList, CallProcessRecordsEvenForEmptyRecordList: DefaultDontCallProcessRecordsForEmptyRecordList,
ParentShardPollIntervalMillis: DefaultParentShardPollIntervalMillis, ParentShardPollIntervalMillis: DefaultParentShardPollIntervalMillis,
ShardSyncIntervalMillis: DefaultShardSyncIntervalMillis, ShardSyncIntervalMillis: DefaultShardSyncIntervalMillis,
CleanupTerminatedShardsBeforeExpiry: DefaultCleanupLeasesUponShardsCompletion, CleanupTerminatedShardsBeforeExpiry: DefaultCleanupLeasesUponShardsCompletion,
TaskBackoffTimeMillis: DefaultTaskBackoffTimeMillis, TaskBackoffTimeMillis: DefaultTaskBackoffTimeMillis,
ValidateSequenceNumberBeforeCheckpointing: DefaultValidateSequenceNumberBeforeCheckpointing, ValidateSequenceNumberBeforeCheckpointing: DefaultValidateSequenceNumberBeforeCheckpointing,
ShutdownGraceMillis: DefaultShutdownGraceMillis, ShutdownGraceMillis: DefaultShutdownGraceMillis,
MaxLeasesForWorker: DefaultMaxLeasesForWorker, MaxLeasesForWorker: DefaultMaxLeasesForWorker,
MaxLeasesToStealAtOneTime: DefaultMaxLeasesToStealAtOneTime, MaxLeasesToStealAtOneTime: DefaultMaxLeasesToStealAtOneTime,
InitialLeaseTableReadCapacity: DefaultInitialLeaseTableReadCapacity, InitialLeaseTableReadCapacity: DefaultInitialLeaseTableReadCapacity,
InitialLeaseTableWriteCapacity: DefaultInitialLeaseTableWriteCapacity, InitialLeaseTableWriteCapacity: DefaultInitialLeaseTableWriteCapacity,
SkipShardSyncAtWorkerInitializationIfLeasesExist: DefaultSkipShardSyncAtStartupIfLeasesExist, SkipShardSyncAtWorkerInitializationIfLeasesExist: DefaultSkipShardSyncAtStartupIfLeasesExist,
EnableLeaseStealing: DefaultEnableLeaseStealing, EnableLeaseStealing: DefaultEnableLeaseStealing,
@ -167,25 +167,24 @@ func (c *KinesisClientLibConfiguration) WithMaxLeasesForWorker(n int) *KinesisCl
return c return c
} }
/* WithIdleTimeBetweenReadsInMillis // WithIdleTimeBetweenReadsInMillis
Controls how long the KCL will sleep if no records are returned from Kinesis // Controls how long the KCL will sleep if no records are returned from Kinesis
//
<p> // <p>
This value is only used when no records are returned; if records are returned, the {@link com.amazonaws.services.kinesis.clientlibrary.lib.worker.ProcessTask} will // This value is only used when no records are returned; if records are returned, the {@link com.amazonaws.services.kinesis.clientlibrary.lib.worker.ProcessTask} will
immediately retrieve the next set of records after the call to // immediately retrieve the next set of records after the call to
{@link com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor#processRecords(ProcessRecordsInput)} // {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor#processRecords(ProcessRecordsInput)}
has returned. Setting this value to high may result in the KCL being unable to catch up. If you are changing this // has returned. Setting this value to high may result in the KCL being unable to catch up. If you are changing this
value it's recommended that you enable {@link #withCallProcessRecordsEvenForEmptyRecordList(boolean)}, and // value it's recommended that you enable {@link #withCallProcessRecordsEvenForEmptyRecordList(boolean)}, and
monitor how far behind the records retrieved are by inspecting // monitor how far behind the records retrieved are by inspecting
{@link com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput#getMillisBehindLatest()}, and the // {@link com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput#getMillisBehindLatest()}, and the
<a href= // <a href=
"http://docs.aws.amazon.com/streams/latest/dev/monitoring-with-cloudwatch.html#kinesis-metrics-stream">CloudWatch // "http://docs.aws.amazon.com/streams/latest/dev/monitoring-with-cloudwatch.html#kinesis-metrics-stream">CloudWatch
Metric: GetRecords.MillisBehindLatest</a> // Metric: GetRecords.MillisBehindLatest</a>
</p> // </p>
//
@param IdleTimeBetweenReadsInMillis: how long to sleep between GetRecords calls when no records are returned. // @param IdleTimeBetweenReadsInMillis: how long to sleep between GetRecords calls when no records are returned.
@return KinesisClientLibConfiguration // @return KinesisClientLibConfiguration
*/
func (c *KinesisClientLibConfiguration) WithIdleTimeBetweenReadsInMillis(idleTimeBetweenReadsInMillis int) *KinesisClientLibConfiguration { func (c *KinesisClientLibConfiguration) WithIdleTimeBetweenReadsInMillis(idleTimeBetweenReadsInMillis int) *KinesisClientLibConfiguration {
checkIsValuePositive("IdleTimeBetweenReadsInMillis", idleTimeBetweenReadsInMillis) checkIsValuePositive("IdleTimeBetweenReadsInMillis", idleTimeBetweenReadsInMillis)
c.IdleTimeBetweenReadsInMillis = idleTimeBetweenReadsInMillis c.IdleTimeBetweenReadsInMillis = idleTimeBetweenReadsInMillis

2
internal/deaggregator/deaggregator.go
View file

@ -93,4 +93,4 @@ func createUserRecord(partitionKeys []string, aggRec *rec.Record, record types.R
PartitionKey: &partitionKey, PartitionKey: &partitionKey,
SequenceNumber: record.SequenceNumber, SequenceNumber: record.SequenceNumber,
} }
} }

4
internal/deaggregator/deaggregator_test.go
View file

@ -61,7 +61,7 @@ func generateKinesisRecord(data []byte) types.Record {
encryptionType := types.EncryptionTypeNone encryptionType := types.EncryptionTypeNone
partitionKey := "1234" partitionKey := "1234"
sequenceNumber := "21269319989900637946712965403778482371" sequenceNumber := "21269319989900637946712965403778482371"
return types.Record { return types.Record{
ApproximateArrivalTimestamp: &currentTime, ApproximateArrivalTimestamp: &currentTime,
Data: data, Data: data,
EncryptionType: encryptionType, EncryptionType: encryptionType,
@ -199,4 +199,4 @@ func TestRecordWithMismatchMd5SumReturnsSingleRecord(t *testing.T) {
// A byte record with an MD5 sum that does not match with the md5.Sum(record) // A byte record with an MD5 sum that does not match with the md5.Sum(record)
// will be marked as a non-aggregate record and return a single record // will be marked as a non-aggregate record and return a single record
assert.Equal(t, 1, len(dars), "Mismatch md5 sum test should return length of 1.") assert.Equal(t, 1, len(dars), "Mismatch md5 sum test should return length of 1.")
} }

2
internal/records/records.pb.go
View file

@ -212,4 +212,4 @@ var fileDescriptor_6ae0159314830e16 = []byte{
0x20, 0x1a, 0x0c, 0xa8, 0x78, 0xbe, 0xd0, 0xe7, 0x26, 0xcd, 0x52, 0x81, 0x08, 0x68, 0x47, 0x25, 0x20, 0x1a, 0x0c, 0xa8, 0x78, 0xbe, 0xd0, 0xe7, 0x26, 0xcd, 0x52, 0x81, 0x08, 0x68, 0x47, 0x25,
0x92, 0x94, 0x28, 0x2b, 0xb4, 0x26, 0x6d, 0x3e, 0x03, 0x00, 0x00, 0xff, 0xff, 0x87, 0x3e, 0x63, 0x92, 0x94, 0x28, 0x2b, 0xb4, 0x26, 0x6d, 0x3e, 0x03, 0x00, 0x00, 0xff, 0xff, 0x87, 0x3e, 0x63,
0x69, 0x7d, 0x01, 0x00, 0x00, 0x69, 0x7d, 0x01, 0x00, 0x00,
} }

2
test/lease_stealing_util_test.go
View file

@ -20,7 +20,7 @@ import (
type LeaseStealingTest struct { type LeaseStealingTest struct {
t *testing.T t *testing.T
config *TestClusterConfig config *TestClusterConfig
cluster *TestCluster cluster *TestCluster
kc *kinesis.Client kc *kinesis.Client
dc *dynamodb.Client dc *dynamodb.Client

14
test/record_publisher_test.go
View file

@ -47,7 +47,7 @@ func NewKinesisClient(t *testing.T, regionName, endpoint string, creds *credenti
t.Logf("Creating Kinesis client") t.Logf("Creating Kinesis client")
resolver := aws.EndpointResolverFunc(func(service, region string) (aws.Endpoint, error) { resolver := aws.EndpointResolverFunc(func(service, region string) (aws.Endpoint, error) {
return aws.Endpoint { return aws.Endpoint{
PartitionID: "aws", PartitionID: "aws",
URL: endpoint, URL: endpoint,
SigningRegion: regionName, SigningRegion: regionName,
@ -73,13 +73,13 @@ func NewKinesisClient(t *testing.T, regionName, endpoint string, creds *credenti
t.Fatalf("Failed in loading Kinesis default config for creating Worker: %+v", err) t.Fatalf("Failed in loading Kinesis default config for creating Worker: %+v", err)
} }
return kinesis.NewFromConfig(cfg) return kinesis.NewFromConfig(cfg)
} }
// NewDynamoDBClient to create a Kinesis Client. // NewDynamoDBClient to create a Kinesis Client.
func NewDynamoDBClient(t *testing.T, regionName, endpoint string, creds *credentials.StaticCredentialsProvider) *dynamodb.Client { func NewDynamoDBClient(t *testing.T, regionName, endpoint string, creds *credentials.StaticCredentialsProvider) *dynamodb.Client {
resolver := aws.EndpointResolverFunc(func(service, region string) (aws.Endpoint, error) { resolver := aws.EndpointResolverFunc(func(service, region string) (aws.Endpoint, error) {
return aws.Endpoint { return aws.Endpoint{
PartitionID: "aws", PartitionID: "aws",
URL: endpoint, URL: endpoint,
SigningRegion: regionName, SigningRegion: regionName,
@ -111,7 +111,7 @@ func continuouslyPublishSomeData(t *testing.T, kc *kinesis.Client) func() {
var shards []types.Shard var shards []types.Shard
var nextToken *string var nextToken *string
for { for {
out, err := kc.ListShards(context.TODO(), &kinesis.ListShardsInput { out, err := kc.ListShards(context.TODO(), &kinesis.ListShardsInput{
StreamName: aws.String(streamName), StreamName: aws.String(streamName),
NextToken: nextToken, NextToken: nextToken,
}) })
@ -185,7 +185,7 @@ func publishSomeData(t *testing.T, kc *kinesis.Client) {
// publishRecord to put a record into Kinesis stream using PutRecord API. // publishRecord to put a record into Kinesis stream using PutRecord API.
func publishRecord(t *testing.T, kc *kinesis.Client, hashKey *string) { func publishRecord(t *testing.T, kc *kinesis.Client, hashKey *string) {
input := &kinesis.PutRecordInput { input := &kinesis.PutRecordInput{
Data: []byte(specstr), Data: []byte(specstr),
StreamName: aws.String(streamName), StreamName: aws.String(streamName),
PartitionKey: aws.String(utils.RandStringBytesMaskImpr(10)), PartitionKey: aws.String(utils.RandStringBytesMaskImpr(10)),
@ -207,7 +207,7 @@ func publishRecords(t *testing.T, kc *kinesis.Client) {
records := make([]types.PutRecordsRequestEntry, 5) records := make([]types.PutRecordsRequestEntry, 5)
for i := 0; i < 5; i++ { for i := 0; i < 5; i++ {
record := types.PutRecordsRequestEntry { record := types.PutRecordsRequestEntry{
Data: []byte(specstr), Data: []byte(specstr),
PartitionKey: aws.String(utils.RandStringBytesMaskImpr(10)), PartitionKey: aws.String(utils.RandStringBytesMaskImpr(10)),
} }
@ -228,7 +228,7 @@ func publishRecords(t *testing.T, kc *kinesis.Client) {
func publishAggregateRecord(t *testing.T, kc *kinesis.Client) { func publishAggregateRecord(t *testing.T, kc *kinesis.Client) {
data := generateAggregateRecord(5, specstr) data := generateAggregateRecord(5, specstr)
// Use random string as partition key to ensure even distribution across shards // Use random string as partition key to ensure even distribution across shards
_, err := kc.PutRecord(context.TODO(), &kinesis.PutRecordInput { _, err := kc.PutRecord(context.TODO(), &kinesis.PutRecordInput{
Data: data, Data: data,
StreamName: aws.String(streamName), StreamName: aws.String(streamName),
PartitionKey: aws.String(utils.RandStringBytesMaskImpr(10)), PartitionKey: aws.String(utils.RandStringBytesMaskImpr(10)),