Skip to content

Commit fd5ee6e

Browse files
shivamsharmashivamsharma
committed
Added exception handling in sending data to dynamo
1 parent e68bdde commit fd5ee6e

File tree

3 files changed

+73
-78
lines changed

3 files changed

+73
-78
lines changed

‎src/main/java/dynamok/commons/Util.java

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
package dynamok.commons;
2+
3+
import com.fasterxml.jackson.core.type.TypeReference;
4+
import com.fasterxml.jackson.databind.ObjectMapper;
5+
import org.apache.kafka.clients.producer.KafkaProducer;
6+
import org.apache.kafka.clients.producer.ProducerConfig;
7+
import org.apache.kafka.common.serialization.StringSerializer;
8+
9+
import java.io.IOException;
10+
import java.util.Map;
11+
import java.util.Properties;
12+
13+
/**
14+
* Project: kafka-connect-dynamodb
15+
* Author: shivamsharma
16+
* Date: 9/22/17.
17+
*/
18+
public class Util {
19+
private static ObjectMapper objectMapper = new ObjectMapper();
20+
21+
public static KafkaProducer<String, String> getKafkaProducer(String broker) {
22+
Properties props = new Properties();
23+
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
24+
props.put(ProducerConfig.CLIENT_ID_CONFIG, "Dynamo Connector Error Pipeline");
25+
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
26+
StringSerializer.class.getName());
27+
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
28+
StringSerializer.class.getName());
29+
return new KafkaProducer<>(props);
30+
}
31+
32+
public static Map<String, Object> jsonToMap(String json) throws IOException {
33+
return objectMapper.readValue(json, new TypeReference<Map<String, String>>() {});
34+
}
35+
}

‎src/main/java/dynamok/sink/ConnectorConfig.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,7 @@ private enum Keys {
8686
.define(Keys.BROKER, ConfigDef.Type.STRING, ConfigDef.NO_DEFAULT_VALUE,
8787
ConfigDef.Importance.HIGH, "Brokers address where Kafka error pipeline will work.")
8888
.define(Keys.ERROR_KAFKA_TOPIC, ConfigDef.Type.STRING, ConfigDef.NO_DEFAULT_VALUE,
89-
ConfigDef.Importance.HIGH, "Error Kafka topic name.");
89+
ConfigDef.Importance.HIGH, "Error kafka topic name.");
9090

9191
final Regions region;
9292
final Password accessKeyId;

‎src/main/java/dynamok/sink/DynamoDbSinkTask.java

Lines changed: 37 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -16,21 +16,21 @@
1616

1717
package dynamok.sink;
1818

19+
import com.amazonaws.auth.AWSStaticCredentialsProvider;
1920
import com.amazonaws.auth.BasicAWSCredentials;
2021
import com.amazonaws.auth.InstanceProfileCredentialsProvider;
22+
import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
2123
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
24+
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
2225
import com.amazonaws.services.dynamodbv2.model.*;
2326
import com.fasterxml.jackson.core.JsonParseException;
24-
import com.fasterxml.jackson.core.type.TypeReference;
2527
import com.fasterxml.jackson.databind.JsonMappingException;
26-
import com.fasterxml.jackson.databind.ObjectMapper;
2728
import dynamok.Version;
29+
import dynamok.commons.Util;
2830
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
2931
import org.apache.kafka.clients.producer.KafkaProducer;
30-
import org.apache.kafka.clients.producer.ProducerConfig;
3132
import org.apache.kafka.clients.producer.ProducerRecord;
3233
import org.apache.kafka.common.TopicPartition;
33-
import org.apache.kafka.common.serialization.StringSerializer;
3434
import org.apache.kafka.connect.data.Schema;
3535
import org.apache.kafka.connect.errors.ConnectException;
3636
import org.apache.kafka.connect.errors.DataException;
@@ -40,6 +40,7 @@
4040
import org.slf4j.Logger;
4141
import org.slf4j.LoggerFactory;
4242

43+
import java.io.IOException;
4344
import java.util.*;
4445

4546

@@ -63,92 +64,62 @@ String topAttributeName(ConnectorConfig config) {
6364
}
6465

6566
private final Logger log = LoggerFactory.getLogger(DynamoDbSinkTask.class);
66-
private ObjectMapper objectMapper = new ObjectMapper();
6767

6868
private ConnectorConfig config;
69-
private AmazonDynamoDBClient client;
69+
private AmazonDynamoDB client;
7070
private KafkaProducer<String, String> producer;
71-
private int remainingRetries;
7271

7372
@Override
7473
public void start(Map<String, String> props) {
7574
config = new ConnectorConfig(props);
76-
producer = getKafkaProducer();
75+
producer = Util.getKafkaProducer(config.broker);
7776

7877
if (config.accessKeyId.value().isEmpty() || config.secretKey.value().isEmpty()) {
79-
client = new AmazonDynamoDBClient(InstanceProfileCredentialsProvider.getInstance());
78+
client = AmazonDynamoDBClientBuilder
79+
.standard()
80+
.withCredentials(InstanceProfileCredentialsProvider.getInstance())
81+
.withRegion(config.region)
82+
.build();
8083
log.debug("AmazonDynamoDBStreamsClient created with DefaultAWSCredentialsProviderChain");
8184
} else {
8285
final BasicAWSCredentials awsCreds = new BasicAWSCredentials(config.accessKeyId.value(), config.secretKey.value());
83-
client = new AmazonDynamoDBClient(awsCreds);
86+
client = AmazonDynamoDBClientBuilder
87+
.standard()
88+
.withCredentials(new AWSStaticCredentialsProvider(awsCreds))
89+
.withRegion(config.region)
90+
.build();
8491
log.debug("AmazonDynamoDBClient created with AWS credentials from connector configuration");
8592
}
86-
87-
client.configureRegion(config.region);
88-
remainingRetries = config.maxRetries;
8993
}
9094

9195
@Override
9296
public void put(Collection<SinkRecord> records) {
9397
if (records.isEmpty()) return;
9498

95-
try {
96-
if (records.size() == 1 || config.batchSize == 1) {
97-
for (final SinkRecord record : records) {
98-
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(config.errorKafkaTopic,
99-
"" + record.key(), record.value().toString());
100-
Map<String, Object> map;
101-
try {
102-
map = objectMapper.readValue(record.value().toString(), new TypeReference<Map<String, String>>() {
103-
});
104-
SinkRecord newRecord = new SinkRecord(record.topic(), record.kafkaPartition(), null,
105-
record.key(), null, map, record.kafkaOffset());
106-
client.putItem(tableName(record), toPutRequest(newRecord).getItem());
107-
} catch (JsonParseException | JsonMappingException e) {
108-
log.error("Exception occurred while converting JSON to Map: {}", record, e);
109-
log.info("Sending to error topic...");
110-
producer.send(producerRecord);
111-
} catch (AmazonDynamoDBException e) {
112-
log.error("Exception in writing into DynamoDB: {}", record, e);
113-
throw e;
114-
} catch (Exception e) {
115-
log.error("Unknown Exception occurred:", e);
116-
producer.send(producerRecord);
117-
}
118-
}
119-
} else {
120-
final Iterator<SinkRecord> recordIterator = records.iterator();
121-
while (recordIterator.hasNext()) {
122-
final Map<String, List<WriteRequest>> writesByTable = toWritesByTable(recordIterator);
123-
final BatchWriteItemResult batchWriteResponse = client.batchWriteItem(new BatchWriteItemRequest(writesByTable));
124-
if (!batchWriteResponse.getUnprocessedItems().isEmpty()) {
125-
throw new UnprocessedItemsException(batchWriteResponse.getUnprocessedItems());
126-
}
127-
}
128-
}
129-
} catch (LimitExceededException | ProvisionedThroughputExceededException e) {
130-
log.debug("Write failed with Limit/Throughput Exceeded exception; backing off");
131-
context.timeout(config.retryBackoffMs);
132-
throw new RetriableException(e);
133-
} catch (AmazonDynamoDBException | UnprocessedItemsException e) {
134-
log.warn("Write failed, remainingRetries={}", remainingRetries, e);
135-
if (remainingRetries == 0) {
136-
ArrayList<SinkRecord> list = new ArrayList<>(records);
137-
log.error("Unable to process this range from: {}\n\t\t\t\t\t\t\tto: {}", list.get(0), list.get(list.size() - 1));
138-
log.info("Writing to error kafka topic: {}", config.errorKafkaTopic);
139-
list.forEach(record -> {
140-
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(config.errorKafkaTopic,
141-
"" + record.key(), record.value().toString());
142-
producer.send(producerRecord);
143-
});
144-
} else {
145-
remainingRetries--;
99+
for (final SinkRecord record : records) {
100+
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(config.errorKafkaTopic,
101+
"" + record.key(), record.value().toString());
102+
try {
103+
SinkRecord newRecord = new SinkRecord(record.topic(), record.kafkaPartition(), null,
104+
record.key(), null, Util.jsonToMap(record.value().toString()), record.kafkaOffset());
105+
client.putItem(tableName(record), toPutRequest(newRecord).getItem());
106+
} catch (JsonParseException | JsonMappingException e) {
107+
log.error("Exception occurred while converting JSON to Map: {}", record, e);
108+
log.info("Sending to error topic...");
109+
producer.send(producerRecord);
110+
} catch (LimitExceededException | ProvisionedThroughputExceededException e) {
111+
log.debug("Write failed with Limit/Throughput Exceeded exception; backing off");
146112
context.timeout(config.retryBackoffMs);
147113
throw new RetriableException(e);
114+
} catch (IOException e) {
115+
log.error("Exception occurred in Json Parsing ", e);
116+
producer.send(producerRecord);
117+
} catch(AmazonDynamoDBException e) {
118+
if (e.getErrorCode().equalsIgnoreCase( "ValidationException")) {
119+
producer.send(producerRecord);
120+
} else throw e;
148121
}
149122
}
150-
151-
remainingRetries = config.maxRetries;
152123
}
153124

154125
private Map<String, List<WriteRequest>> toWritesByTable(Iterator<SinkRecord> recordIterator) {
@@ -219,15 +190,4 @@ public String version() {
219190
return Version.get();
220191
}
221192

222-
private KafkaProducer<String, String> getKafkaProducer() {
223-
Properties props = new Properties();
224-
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.broker);
225-
props.put(ProducerConfig.CLIENT_ID_CONFIG, "Dynamo Sink Connector Error Pipeline");
226-
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
227-
StringSerializer.class.getName());
228-
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
229-
StringSerializer.class.getName());
230-
return new KafkaProducer<>(props);
231-
}
232-
233193
}

0 commit comments

Comments
 (0)