1+ /*
2+ * Copyright 2013-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+ *
4+ * Licensed under the Amazon Software License (the "License").
5+ * You may not use this file except in compliance with the License.
6+ * A copy of the License is located at
7+ *
8+ * http://aws.amazon.com/asl/
9+ *
10+ * or in the "license" file accompanying this file. This file is distributed
11+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
12+ * express or implied. See the License for the specific language governing
13+ * permissions and limitations under the License.
14+ */
15+ package com .sumologic .kinesis ;
16+
17+ import java .io .IOException ;
18+ import java .util .ArrayList ;
19+ import java .util .Collection ;
20+ import java .util .List ;
21+
22+ import org .apache .commons .logging .Log ;
23+ import org .apache .commons .logging .LogFactory ;
24+
25+ import com .amazonaws .services .kinesis .clientlibrary .exceptions .InvalidStateException ;
26+ import com .amazonaws .services .kinesis .clientlibrary .exceptions .KinesisClientLibDependencyException ;
27+ import com .amazonaws .services .kinesis .clientlibrary .exceptions .ShutdownException ;
28+ import com .amazonaws .services .kinesis .clientlibrary .exceptions .ThrottlingException ;
29+ import com .amazonaws .services .kinesis .clientlibrary .interfaces .IRecordProcessor ;
30+ import com .amazonaws .services .kinesis .clientlibrary .interfaces .IRecordProcessorCheckpointer ;
31+ import com .amazonaws .services .kinesis .clientlibrary .types .ShutdownReason ;
32+ import com .amazonaws .services .kinesis .connectors .KinesisConnectorConfiguration ;
33+ import com .amazonaws .services .kinesis .connectors .UnmodifiableBuffer ;
34+ import com .amazonaws .services .kinesis .connectors .interfaces .IBuffer ;
35+ import com .amazonaws .services .kinesis .connectors .interfaces .ICollectionTransformer ;
36+ import com .amazonaws .services .kinesis .connectors .interfaces .IEmitter ;
37+ import com .amazonaws .services .kinesis .connectors .interfaces .IFilter ;
38+ import com .amazonaws .services .kinesis .connectors .interfaces .ITransformer ;
39+ import com .amazonaws .services .kinesis .connectors .interfaces .ITransformerBase ;
40+ import com .amazonaws .services .kinesis .model .Record ;
41+
42+ /**
43+ * This is the base class for any KinesisConnector. It is configured by a constructor that takes in
44+ * as parameters implementations of the IBuffer, ITransformer, and IEmitter dependencies defined in
45+ * a IKinesisConnectorPipeline. It is typed to match the class that records are transformed into for
46+ * filtering and manipulation. This class is produced by a KinesisConnectorRecordProcessorFactory.
47+ * <p>
48+ * When a Worker calls processRecords() on this class, the pipeline is used in the following way:
49+ * <ol>
50+ * <li>Records are transformed into the corresponding data model (parameter type T) via the ITransformer.</li>
51+ * <li>Transformed records are passed to the IBuffer.consumeRecord() method, which may optionally filter based on the
52+ * IFilter in the pipeline.</li>
53+ * <li>When the buffer is full (IBuffer.shouldFlush() returns true), records are transformed with the ITransformer to
54+ * the output type (parameter type U) and a call is made to IEmitter.emit(). IEmitter.emit() returning an empty list is
55+ * considered a success, so the record processor will checkpoint and emit will not be retried. Non-empty return values
56+ * will result in additional calls to emit with failed records as the unprocessed list until the retry limit is reached.
57+ * Upon exceeding the retry limit or an exception being thrown, the IEmitter.fail() method will be called with the
58+ * unprocessed records.</li>
59+ * <li>When the shutdown() method of this class is invoked, a call is made to the IEmitter.shutdown() method which
60+ * should close any existing client connections.</li>
61+ * </ol>
62+ *
63+ */
64+ public class KinesisConnectorRecordProcessor <T , U > implements IRecordProcessor {
65+
66+ private final IEmitter <U > emitter ;
67+ private final ITransformerBase <T , U > transformer ;
68+ private final IFilter <T > filter ;
69+ private final IBuffer <T > buffer ;
70+ private final int retryLimit ;
71+ private final long backoffInterval ;
72+ private boolean isShutdown = false ;
73+
74+ private static final Log LOG = LogFactory .getLog (KinesisConnectorRecordProcessor .class );
75+
76+ private String shardId ;
77+
78+ public KinesisConnectorRecordProcessor (IBuffer <T > buffer ,
79+ IFilter <T > filter ,
80+ IEmitter <U > emitter ,
81+ ITransformerBase <T , U > transformer ,
82+ KinesisConnectorConfiguration configuration ) {
83+ if (buffer == null || filter == null || emitter == null || transformer == null ) {
84+ throw new IllegalArgumentException ("buffer, filter, emitter, and transformer must not be null" );
85+ }
86+ this .buffer = buffer ;
87+ this .filter = filter ;
88+ this .emitter = emitter ;
89+ this .transformer = transformer ;
90+ // Limit must be greater than zero
91+ if (configuration .RETRY_LIMIT <= 0 ) {
92+ retryLimit = 1 ;
93+ } else {
94+ retryLimit = configuration .RETRY_LIMIT ;
95+ }
96+ this .backoffInterval = configuration .BACKOFF_INTERVAL ;
97+ }
98+
99+ @ Override
100+ public void initialize (String shardId ) {
101+ this .shardId = shardId ;
102+ }
103+
104+ @ Override
105+ public void processRecords (List <Record > records , IRecordProcessorCheckpointer checkpointer ) {
106+ // Note: This method will be called even for empty record lists. This is needed for checking the buffer time
107+ // threshold.
108+ if (isShutdown ) {
109+ LOG .warn ("processRecords called on shutdown record processor for shardId: " + shardId );
110+ return ;
111+ }
112+ if (shardId == null ) {
113+ throw new IllegalStateException ("Record processor not initialized" );
114+ }
115+
116+ // Transform each Amazon Kinesis Record and add the result to the buffer
117+ for (Record record : records ) {
118+ try {
119+ if (transformer instanceof ITransformer ) {
120+ ITransformer <T , U > singleTransformer = (ITransformer <T , U >) transformer ;
121+ filterAndBufferRecord (singleTransformer .toClass (record ), record );
122+ } else if (transformer instanceof ICollectionTransformer ) {
123+ ICollectionTransformer <T , U > listTransformer = (ICollectionTransformer <T , U >) transformer ;
124+ Collection <T > transformedRecords = listTransformer .toClass (record );
125+ for (T transformedRecord : transformedRecords ) {
126+ filterAndBufferRecord (transformedRecord , record );
127+ }
128+ } else {
129+ throw new RuntimeException ("Transformer must implement ITransformer or ICollectionTransformer" );
130+ }
131+ } catch (IOException e ) {
132+ LOG .error (e );
133+ }
134+ }
135+
136+ if (buffer .shouldFlush ()) {
137+ List <U > emitItems = transformToOutput (buffer .getRecords ());
138+ emit (checkpointer , emitItems );
139+ }
140+ }
141+
142+ private void filterAndBufferRecord (T transformedRecord , Record record ) {
143+ if (filter .keepRecord (transformedRecord )) {
144+ buffer .consumeRecord (transformedRecord , record .getData ().array ().length , record .getSequenceNumber ());
145+ }
146+ }
147+
148+ private List <U > transformToOutput (List <T > items ) {
149+ List <U > emitItems = new ArrayList <U >();
150+ for (T item : items ) {
151+ try {
152+ emitItems .add (transformer .fromClass (item ));
153+ } catch (IOException e ) {
154+ LOG .error ("Failed to transform record " + item + " to output type" , e );
155+ }
156+ }
157+ return emitItems ;
158+ }
159+
160+ private void emit (IRecordProcessorCheckpointer checkpointer , List <U > emitItems ) {
161+ List <U > unprocessed = new ArrayList <U >(emitItems );
162+ try {
163+ for (int numTries = 0 ; numTries < retryLimit ; numTries ++) {
164+ unprocessed = emitter .emit (new UnmodifiableBuffer <U >(buffer , unprocessed ));
165+ if (unprocessed .isEmpty ()) {
166+ break ;
167+ }
168+ try {
169+ Thread .sleep (backoffInterval );
170+ } catch (InterruptedException e ) {
171+ }
172+ }
173+ if (!unprocessed .isEmpty ()) {
174+ emitter .fail (unprocessed );
175+ }
176+ final String lastSequenceNumberProcessed = buffer .getLastSequenceNumber ();
177+ buffer .clear ();
178+ // checkpoint once all the records have been consumed
179+ if (lastSequenceNumberProcessed != null && unprocessed .isEmpty ()) {
180+ checkpointer .checkpoint (lastSequenceNumberProcessed );
181+ }
182+ } catch (IOException | KinesisClientLibDependencyException | InvalidStateException | ThrottlingException
183+ | ShutdownException e ) {
184+ LOG .error (e );
185+ emitter .fail (unprocessed );
186+ }
187+ }
188+
189+ @ Override
190+ public void shutdown (IRecordProcessorCheckpointer checkpointer , ShutdownReason reason ) {
191+ LOG .info ("Shutting down record processor with shardId: " + shardId + " with reason " + reason );
192+ if (isShutdown ) {
193+ LOG .warn ("Record processor for shardId: " + shardId + " has been shutdown multiple times." );
194+ return ;
195+ }
196+ switch (reason ) {
197+ case TERMINATE :
198+ emit (checkpointer , transformToOutput (buffer .getRecords ()));
199+ try {
200+ checkpointer .checkpoint ();
201+ } catch (KinesisClientLibDependencyException | InvalidStateException | ThrottlingException | ShutdownException e ) {
202+ LOG .error (e );
203+ }
204+ break ;
205+ case ZOMBIE :
206+ break ;
207+ default :
208+ throw new IllegalStateException ("invalid shutdown reason" );
209+ }
210+ emitter .shutdown ();
211+ isShutdown = true ;
212+ }
213+
214+ }
0 commit comments