Belated changes for KCL v3 migration #56
Conversation
rtyley
force-pushed
the
changes-for-kcl-v3-migration
branch
4 times, most recently
from
4b5f8be to
2b5535f
Compare
rtyley
force-pushed
the
changes-for-kcl-v3-migration
branch
from
2b5535f to
4e2ea55
Compare
|
|
||
| import scala.concurrent.duration._ | ||
|
|
||
| class ContentApiFirehoseConsumer( | ||
| val kinesisStreamReaderConfig: KinesisStreamReaderConfig, | ||
| override val credentialsProvider: AwsCredentialsProvider, | ||
| val streamListener: StreamListener, | ||
| val filterProductionMonitoring: Boolean = false) extends KinesisStreamReader { | ||
| val filterProductionMonitoring: Boolean = false, | ||
| val clientVersionCompatibility: CoordinatorConfig.ClientVersionConfig // see https://github.com/guardian/content-api-firehose-client/pull/56 |
There was a problem hiding this comment.
Migration Step 3: Migration-related configuration
I haven't provided a default value for clientVersionCompatibility here, in order to force consideration in projects that use KCL in more than one place - they need to be aware that by taking this library upgrade, they're using KCL v3 everywhere, and this ClientVersionConfig setting should be made on the coordinatorConfig of every KCL client that's in the project.
Recommended value of clientVersionCompatibility
Use CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X, for at least one deploy, until all workers are emitting WorkerMetricStats that show they have upgraded.
| val ongoingCalls = ongoingProcessRecordCallsPhaser.getUnarrivedParties - 1 | ||
| if (ongoingCalls > 0) | ||
| logger.info(s"Shutdown requested but waiting for processRecords() to complete - ongoingCalls=$ongoingCalls") | ||
|
|
||
| // Ensure that all records we've received have been processed before we call checkpoint and exit | ||
| Try(ongoingProcessRecordCallsPhaser.awaitAdvanceInterruptibly(ongoingProcessRecordCallsPhaser.arrive(), 10, SECONDS)).recover { | ||
| case e: TimeoutException => logger.error("Timeout while waiting for processRecords() to complete", e) | ||
| } |
There was a problem hiding this comment.
Migration Step 4: Graceful lease handoff
If you are performing any asynchronous processing, make sure that all delivered records to the downstream were processed before invoking checkpointing.
I wasn't sure how to interpret this, but I think it is possible for a call to processRecords() (invoked by software.amazon.kinesis.lifecycle.ProcessTask) to be ongoing while shutdownRequested() is called (invoked by software.amazon.kinesis.lifecycle.ShutdownNotificationTask).
If that's the case, we want to make sure we've finished our processing work in processRecords() before we checkpoint in shutdownRequested(), while being conscious that it could be occurring in a different thread.
See: