소비자를 KCL 1.x에서 2.x로 KCL 마이그레이션 - Amazon Kinesis Data Streams

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

소비자를 KCL 1.x에서 2.x로 KCL 마이그레이션

이 항목에서는 Kinesis 클라이언트 라이브러리 버전 1.x와 2.x 버전 () 간의 차이점을 설명합니다. KCL 또한 소비자를 버전 1.x에서 버전 2.x로 마이그레이션하는 방법도 보여줍니다. KCL 클라이언트 마이그레이션 후 마지막 체크포인트 위치에서 처리 기록을 시작합니다.

버전 2.0에서는 다음과 같은 인터페이스 변경 사항이 KCL 도입되었습니다.

KCL인터페이스 변경
KCL1.x 인터페이스 KCL2.0 인터페이스
com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor software.amazon.kinesis.processor.ShardRecordProcessor
com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory software.amazon.kinesis.processor.ShardRecordProcessorFactory
com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware software.amazon.kinesis.processor.ShardRecordProcessor에 포함됨

레코드 프로세서 마이그레이션

다음 예제는 KCL 1.x용으로 구현된 레코드 프로세서를 보여줍니다.

package com.amazonaws.kcl; import com.amazonaws.services.kinesis.clientlibrary.exceptions.InvalidStateException; import com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException; import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer; import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor; import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware; import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason; import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput; import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput; import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput; public class TestRecordProcessor implements IRecordProcessor, IShutdownNotificationAware { @Override public void initialize(InitializationInput initializationInput) { // // Setup record processor // } @Override public void processRecords(ProcessRecordsInput processRecordsInput) { // // Process records, and possibly checkpoint // } @Override public void shutdown(ShutdownInput shutdownInput) { if (shutdownInput.getShutdownReason() == ShutdownReason.TERMINATE) { try { shutdownInput.getCheckpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { throw new RuntimeException(e); } } } @Override public void shutdownRequested(IRecordProcessorCheckpointer checkpointer) { try { checkpointer.checkpoint(); } catch (ShutdownException | InvalidStateException e) { // // Swallow exception // e.printStackTrace(); } } }
레코드 프로세서 클래스를 마이그레이션하려면
  1. 다음과 같이 com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorcom.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware의 인터페이스를 software.amazon.kinesis.processor.ShardRecordProcessor로 변경합니다.

    // import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor; // import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware; import software.amazon.kinesis.processor.ShardRecordProcessor; // public class TestRecordProcessor implements IRecordProcessor, IShutdownNotificationAware { public class TestRecordProcessor implements ShardRecordProcessor {
  2. importinitialize 메서드의 processRecords 문을 업데이트합니다.

    // import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput; import software.amazon.kinesis.lifecycle.events.InitializationInput; //import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput; import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput;
  3. shutdown 메서드를 새 메서드인 leaseLost, shardEnded, shutdownRequested으로 바꿉니다.

    // @Override // public void shutdownRequested(IRecordProcessorCheckpointer checkpointer) { // // // // This is moved to shardEnded(...) // // // try { // checkpointer.checkpoint(); // } catch (ShutdownException | InvalidStateException e) { // // // // Swallow exception // // // e.printStackTrace(); // } // } @Override public void leaseLost(LeaseLostInput leaseLostInput) { } @Override public void shardEnded(ShardEndedInput shardEndedInput) { try { shardEndedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { // // Swallow the exception // e.printStackTrace(); } } // @Override // public void shutdownRequested(IRecordProcessorCheckpointer checkpointer) { // // // // This is moved to shutdownRequested(ShutdownReauestedInput) // // // try { // checkpointer.checkpoint(); // } catch (ShutdownException | InvalidStateException e) { // // // // Swallow exception // // // e.printStackTrace(); // } // } @Override public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) { try { shutdownRequestedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { // // Swallow the exception // e.printStackTrace(); } }

다음은 업데이트된 버전의 레코드 프로세서 클래스입니다.

package com.amazonaws.kcl; import software.amazon.kinesis.exceptions.InvalidStateException; import software.amazon.kinesis.exceptions.ShutdownException; import software.amazon.kinesis.lifecycle.events.InitializationInput; import software.amazon.kinesis.lifecycle.events.LeaseLostInput; import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput; import software.amazon.kinesis.lifecycle.events.ShardEndedInput; import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput; import software.amazon.kinesis.processor.ShardRecordProcessor; public class TestRecordProcessor implements ShardRecordProcessor { @Override public void initialize(InitializationInput initializationInput) { } @Override public void processRecords(ProcessRecordsInput processRecordsInput) { } @Override public void leaseLost(LeaseLostInput leaseLostInput) { } @Override public void shardEnded(ShardEndedInput shardEndedInput) { try { shardEndedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { // // Swallow the exception // e.printStackTrace(); } } @Override public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) { try { shutdownRequestedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { // // Swallow the exception // e.printStackTrace(); } } }

레코드 프로세서 팩토리 마이그레이션

레코드 프로세스 팩토리는 리스가 필요할 경우 레코드 프로세서 생성을 담당합니다. 다음은 KCL 1.x 팩토리의 예입니다.

package com.amazonaws.kcl; import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor; import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory; public class TestRecordProcessorFactory implements IRecordProcessorFactory { @Override public IRecordProcessor createProcessor() { return new TestRecordProcessor(); } }
레코드 프로세서 팩토리를 마이그레이션하려면
  1. 구현된 인터페이스를 다음과 같이 com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory에서 software.amazon.kinesis.processor.ShardRecordProcessorFactory로 변경합니다.

    // import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor; import software.amazon.kinesis.processor.ShardRecordProcessor; // import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory; import software.amazon.kinesis.processor.ShardRecordProcessorFactory; // public class TestRecordProcessorFactory implements IRecordProcessorFactory { public class TestRecordProcessorFactory implements ShardRecordProcessorFactory {
  2. createProcessor에 대한 반환 서명을 변경합니다.

    // public IRecordProcessor createProcessor() { public ShardRecordProcessor shardRecordProcessor() {

다음은 2.0의 레코드 프로세서 팩토리의 예입니다.

package com.amazonaws.kcl; import software.amazon.kinesis.processor.ShardRecordProcessor; import software.amazon.kinesis.processor.ShardRecordProcessorFactory; public class TestRecordProcessorFactory implements ShardRecordProcessorFactory { @Override public ShardRecordProcessor shardRecordProcessor() { return new TestRecordProcessor(); } }

작업자 마이그레이션

버전 KCL Scheduler 2.0에서는 라는 새 클래스가 클래스를 대체합니다Worker. 다음은 KCL 1.x 워커의 예제입니다.

final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...) final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory(); final Worker worker = new Worker.Builder() .recordProcessorFactory(recordProcessorFactory) .config(config) .build();
작업자를 마이그레이션하려면
  1. Worker 클래스에 대한 import 문을 SchedulerConfigsBuilder 클래스에 대한 가져오기 문으로 변경합니다.

    // import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker; import software.amazon.kinesis.coordinator.Scheduler; import software.amazon.kinesis.common.ConfigsBuilder;
  2. 다음 예와 같이 ConfigsBuilderScheduler를 생성합니다.

    KinesisClientUtil을 사용하여 KinesisAsyncClient를 만들고 KinesisAsyncClientmaxConcurrency를 구성하는 것이 좋습니다.

    중요

    Amazon Kinesis Client는 KinesisAsyncClient를 구성하여 KinesisAsyncClient의 전체 임대 수에 더해 추가 사용량까지 허용할 수 있을 만큼 maxConcurrency를 충분히 높이지 않을 경우 지연 시간이 크게 증가할 수 있습니다.

    import java.util.UUID; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import software.amazon.kinesis.common.ConfigsBuilder; import software.amazon.kinesis.common.KinesisClientUtil; import software.amazon.kinesis.coordinator.Scheduler; ... Region region = Region.AP_NORTHEAST_2; KinesisAsyncClient kinesisClient = KinesisClientUtil.createKinesisAsyncClient(KinesisAsyncClient.builder().region(region)); DynamoDbAsyncClient dynamoClient = DynamoDbAsyncClient.builder().region(region).build(); CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().region(region).build(); ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, applicationName, kinesisClient, dynamoClient, cloudWatchClient, UUID.randomUUID().toString(), new SampleRecordProcessorFactory()); Scheduler scheduler = new Scheduler( configsBuilder.checkpointConfig(), configsBuilder.coordinatorConfig(), configsBuilder.leaseManagementConfig(), configsBuilder.lifecycleConfig(), configsBuilder.metricsConfig(), configsBuilder.processorConfig(), configsBuilder.retrievalConfig() );

Amazon Kinesis 클라이언트 구성

Kinesis Client Library 2.0 릴리스에서는 클라이언트 구성이 단일 구성 클래스(KinesisClientLibConfiguration)에서 6개 구성 클래스로 이동했습니다. 다음 표에 마이그레이션에 대한 설명이 나와 있습니다.

구성 필드와 새 클래스
원래 필드 새 구성 클래스 설명
applicationName ConfigsBuilder 이 KCL 애플리케이션의 이름입니다. tableNameconsumerName의 기본값으로 사용됩니다.
tableName ConfigsBuilder Amazon DynamoDB 리스 테이블에 사용된 테이블 이름 재정의를 허용합니다.
streamName ConfigsBuilder 이 애플리케이션이 레코드를 처리하는 스트림의 이름입니다.
kinesisEndpoint ConfigsBuilder 이 옵션은 제거되었습니다. 클라이언트 구성 제거를 참조하십시오.
dynamoDBEndpoint ConfigsBuilder 이 옵션은 제거되었습니다. 클라이언트 구성 제거를 참조하십시오.
initialPositionInStreamExtended RetrievalConfig 애플리케이션이 처음 실행될 때부터 레코드를 가져오기 시작하는 샤드 내 위치입니다. KCL
kinesisCredentialsProvider ConfigsBuilder 이 옵션은 제거되었습니다. 클라이언트 구성 제거를 참조하십시오.
dynamoDBCredentialsProvider ConfigsBuilder 이 옵션은 제거되었습니다. 클라이언트 구성 제거를 참조하십시오.
cloudWatchCredentialsProvider ConfigsBuilder 이 옵션은 제거되었습니다. 클라이언트 구성 제거를 참조하십시오.
failoverTimeMillis LeaseManagementConfig 리스 소유자가 실패했다고 간주하기 전에 전달해야 하는 시간(밀리초)입니다.
workerIdentifier ConfigsBuilder 이 애플리케이션 프로세서 인스턴스화를 나타내는 고유 식별자입니다. 고유해야 합니다.
shardSyncIntervalMillis LeaseManagementConfig 샤드 sync 호출 사이의 시간.
maxRecords PollingConfig Kinesis가 반환하는 최대 레코드 수를 설정할 수 있습니다.
idleTimeBetweenReadsInMillis CoordinatorConfig 이 옵션은 제거되었습니다. 유휴 시간 제거를 참조하십시오
callProcessRecordsEvenForEmptyRecordList ProcessorConfig 설정하면, Kinesis에서 제공된 레코드가 없는 경우에도 레코드 프로세서가 직접적으로 호출됩니다.
parentShardPollIntervalMillis CoordinatorConfig 상위 샤드가 완료되었는지를 확인하기 위해 레코드 프로세서가 폴링을 수행해야 하는 간격입니다.
cleanupLeasesUponShardCompletion LeaseManagementConfig 설정하면, 하위 리스에서 처리를 시작하자마자 리스가 제거됩니다.
ignoreUnexpectedChildShards LeaseManagementConfig 설정하면, 진행 중인 샤드가 있는 하위 샤드가 무시됩니다. 이는 주로 DynamoDB Streams용입니다.
kinesisClientConfig ConfigsBuilder 이 옵션은 제거되었습니다. 클라이언트 구성 제거를 참조하십시오.
dynamoDBClientConfig ConfigsBuilder 이 옵션은 제거되었습니다. 클라이언트 구성 제거를 참조하십시오.
cloudWatchClientConfig ConfigsBuilder 이 옵션은 제거되었습니다. 클라이언트 구성 제거를 참조하십시오.
taskBackoffTimeMillis LifecycleConfig 실패한 작업을 재시도하기 위한 대기 시간.
metricsBufferTimeMillis MetricsConfig 메트릭 게시를 제어합니다 CloudWatch .
metricsMaxQueueSize MetricsConfig CloudWatch 메트릭 게시를 제어합니다.
metricsLevel MetricsConfig CloudWatch 메트릭 게시를 제어합니다.
metricsEnabledDimensions MetricsConfig CloudWatch 메트릭 게시를 제어합니다.
validateSequenceNumberBeforeCheckpointing CheckpointConfig 이 옵션은 제거되었습니다. 체크포인트 시퀀스 번호 검증을 참조하십시오.
regionName ConfigsBuilder 이 옵션은 제거되었습니다. 클라이언트 구성 제거를 참조하십시오.
maxLeasesForWorker LeaseManagementConfig 애플리케이션의 한 인스턴스가 허용해야 하는 최대 리스 수입니다.
maxLeasesToStealAtOneTime LeaseManagementConfig 애플리케이션이 한 번에 스틸을 시도해야 하는 최대 리스 수입니다.
initialLeaseTableReadCapacity LeaseManagementConfig Kinesis 클라이언트 라이브러리가 새 DynamoDB 임대 테이블을 생성해야 하는 경우 사용되는 DynamoDB IOPs 읽기입니다.
initialLeaseTableWriteCapacity LeaseManagementConfig Kinesis 클라이언트 라이브러리가 새 DynamoDB 임대 테이블을 생성해야 하는 경우 사용되는 DynamoDB IOPs 읽기입니다.
initialPositionInStreamExtended LeaseManagementConfig 애플리케이션이 읽기를 시작해야 하는 스트림 내 초기 위치입니다. 최초 리스 생성 시에만 사용됩니다.
skipShardSyncAtWorkerInitializationIfLeasesExist CoordinatorConfig 리스 테이블에 기존 리스가 있는 경우 샤드 데이터 동기화를 비활성화합니다. TODO KinesisEco: -438
shardPrioritization CoordinatorConfig 사용할 샤드 우선 순위.
shutdownGraceMillis N/A 이 옵션은 제거되었습니다. 제거를 참조하십시오 MultiLang .
timeoutInSeconds N/A 이 옵션은 제거되었습니다. MultiLang 삭제를 참조하십시오.
retryGetRecordsInSeconds PollingConfig 실패 GetRecords 시도 사이의 지연 시간을 구성합니다.
maxGetRecordsThreadPool PollingConfig 사용된 스레드 풀 크기. GetRecords
maxLeaseRenewalThreads LeaseManagementConfig 리스 갱신 스레드 풀의 크기를 제어합니다. 애플리케이션에서 사용할 수 있는 리스가 많을수록 이 풀의 크기가 커야 합니다.
recordsFetcherFactory PollingConfig 스트림에서 검색하는 페처를 생성하는 데 사용되는 팩토리를 교체할 수 있습니다.
logWarningForTaskAfterMillis LifecycleConfig 작업이 완료되지 않은 경우 얼마나 대기한 후 경고가 기록될지를 지정합니다.
listShardsBackoffTimeInMillis RetrievalConfig 실패 발생 시 ListShards 호출 간에 대기하는 시간(밀리초)입니다.
maxListShardsRetryAttempts RetrievalConfig 포기하기 전에 ListShards가 재시도하는 최대 횟수입니다.

유휴 시간 제거

1.x 버전에서는 다음과 KCL 같은 두 가지 idleTimeBetweenReadsInMillis 수량에 해당합니다.

  • 작업 디스패칭 확인 간의 시간. 작업 간의 이 시간은 이제 CoordinatorConfig#shardConsumerDispatchPollIntervalMillis를 설정하여 구성할 수 있습니다.

  • Kinesis Data Streams에서 반환하는 레코드가 없는 경우 절전 모드로 들어가는 시간. 버전 2.0에서는 향상된 팬아웃 레코드가 해당 검색자로부터 푸시됩니다. 샤드 소비자에 대한 작업은 푸시된 요청이 도착한 경우에만 발생합니다.

클라이언트 구성 제거

버전 2.0에서는 더 KCL 이상 클라이언트를 생성하지 않습니다. 이제 사용자가 유효한 클라이언트를 제공해야 합니다. 따라서 클라이언트 생성을 제어하던 모든 구성 파라미터는 삭제되었습니다. 이러한 파라미터가 필요한 경우 ConfigsBuilder에 클라이언트를 제공하기 전에 클라이언트에서 설정할 수 있습니다.

제거된 필드 동일 구성
kinesisEndpoint 기본 엔드포인트로 구성:KinesisAsyncClient.builder().endpointOverride(URI.create("https://<kinesis endpoint>")).build(). SDK KinesisAsyncClient
dynamoDBEndpoint 선호 엔드포인트로 구성:DynamoDbAsyncClient.builder().endpointOverride(URI.create("https://<dynamodb endpoint>")).build(). SDK DynamoDbAsyncClient
kinesisClientConfig 필요한 SDK KinesisAsyncClient 구성으로 구성:KinesisAsyncClient.builder().overrideConfiguration(<your configuration>).build().
dynamoDBClientConfig 필요한 SDK DynamoDbAsyncClient 구성으로 구성:DynamoDbAsyncClient.builder().overrideConfiguration(<your configuration>).build().
cloudWatchClientConfig 필요한 SDK CloudWatchAsyncClient 구성으로 구성:CloudWatchAsyncClient.builder().overrideConfiguration(<your configuration>).build().
regionName 원하는 지역으로 SDK 구성하십시오. 이는 모든 SDK 클라이언트에서 동일합니다. 예: KinesisAsyncClient.builder().region(Region.US_WEST_2).build().