446
}
447
448
>
func scaleStateToReadCount(scaleState *persistencespb.PartitionScaleState) int32 {
scale_manager.go
449
>
return max(scaleState.GetTarget(), bitSet(scaleState.GetBacklogState()).len())
450
>
}
451
452
func scaleStateToInfo(
453
scaleState *persistencespb.PartitionScaleState,
454
settings dynamicconfig.PartitionScaleManagerSettings,
456
>
// note if scaleState == nil, read and write will both be 0
457
>
read := scaleStateToReadCount(scaleState)
458
>
allowedShrink := max(
459
>
1,
460
>
min(
461
>
int32(float32(read)*settings.ShrinkRatio),
462
>
settings.ShrinkDelta,
463
>
),
464
>
)
465
>
write := max(
466
>
scaleState.GetTarget(),
467
>
read-allowedShrink,
468
>
)
469
>
return &taskqueuespb.PartitionScaleInfo{
470
>
Read: read,
471
>
Write: write,
472
>
BacklogCounts: scaleState.GetBacklogCounts(),
473
>
BacklogCap: scaleState.GetBacklogCap(),
474
>
Version: scaleState.GetTargetVersion(),
475
>
}
476
>
}