navina commented on code in PR #8609:
URL: https://github.com/apache/pinot/pull/8609#discussion_r861180460
##########
pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/server/KinesisDataProducer.java:
##########
@@ -54,42 +71,56 @@ public void init(Properties props) {
.credentialsProvider(getLocalAWSCredentials(props))
.httpClientBuilder(new
ApacheSdkHttpService().createHttpClientBuilder());
} else {
- kinesisClientBuilder =
-
KinesisClient.builder().region(Region.of(props.getProperty(REGION)))
- .credentialsProvider(DefaultCredentialsProvider.create())
- .httpClientBuilder(new
ApacheSdkHttpService().createHttpClientBuilder());
+ kinesisClientBuilder =
KinesisClient.builder().region(Region.of(props.getProperty(REGION)))
+ .credentialsProvider(DefaultCredentialsProvider.create())
+ .httpClientBuilder(new
ApacheSdkHttpService().createHttpClientBuilder());
}
if (props.containsKey(ENDPOINT)) {
String kinesisEndpoint = props.getProperty(ENDPOINT, DEFAULT_ENDPOINT);
try {
kinesisClientBuilder = kinesisClientBuilder.endpointOverride(new
URI(kinesisEndpoint));
} catch (URISyntaxException e) {
- throw new IllegalArgumentException("URI syntax is not correctly
specified for endpoint: "
- + kinesisEndpoint, e);
+ throw new IllegalArgumentException("URI syntax is not correctly
specified for endpoint: " + kinesisEndpoint,
+ e);
}
}
_kinesisClient = kinesisClientBuilder.build();
+
+ int numRetries = Integer.parseInt(props.getProperty(NUM_RETRIES,
DEFAULT_NUM_RETRIES));
+ long retryDelayMs = Long.parseLong(props.getProperty(RETRY_DELAY_MILLIS,
DEFAULT_RETRY_DELAY_MILLIS));
+ _retryPolicy = new FixedDelayRetryPolicy(numRetries, retryDelayMs);
} catch (Exception e) {
_kinesisClient = null;
}
}
@Override
public void produce(String topic, byte[] payload) {
- PutRecordRequest putRecordRequest =
-
PutRecordRequest.builder().streamName(topic).data(SdkBytes.fromByteArray(payload))
- .partitionKey(UUID.randomUUID().toString()).build();
- PutRecordResponse putRecordResponse =
_kinesisClient.putRecord(putRecordRequest);
+ try {
+ _retryPolicy.attempt(() -> putRecord(topic, null, payload));
+ } catch (Exception e) {
+ LOGGER.error("Retries exhausted while pushing record in stream {}",
topic);
+ }
}
@Override
public void produce(String topic, byte[] key, byte[] payload) {
- PutRecordRequest putRecordRequest =
-
PutRecordRequest.builder().streamName(topic).data(SdkBytes.fromByteArray(payload)).partitionKey(new
String(key))
- .build();
- PutRecordResponse putRecordResponse =
_kinesisClient.putRecord(putRecordRequest);
+ try {
+ _retryPolicy.attempt(() -> putRecord(topic, key, payload));
Review Comment:
how is exception thrown in the lambda handled? looks like putRecord already
has a try-catch within the method. When it returns false, does this
`_retryPolicy.attempt` throw an exception?
##########
pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/server/KinesisDataProducer.java:
##########
@@ -54,42 +71,56 @@ public void init(Properties props) {
.credentialsProvider(getLocalAWSCredentials(props))
.httpClientBuilder(new
ApacheSdkHttpService().createHttpClientBuilder());
} else {
- kinesisClientBuilder =
-
KinesisClient.builder().region(Region.of(props.getProperty(REGION)))
- .credentialsProvider(DefaultCredentialsProvider.create())
- .httpClientBuilder(new
ApacheSdkHttpService().createHttpClientBuilder());
+ kinesisClientBuilder =
KinesisClient.builder().region(Region.of(props.getProperty(REGION)))
+ .credentialsProvider(DefaultCredentialsProvider.create())
+ .httpClientBuilder(new
ApacheSdkHttpService().createHttpClientBuilder());
}
if (props.containsKey(ENDPOINT)) {
String kinesisEndpoint = props.getProperty(ENDPOINT, DEFAULT_ENDPOINT);
try {
kinesisClientBuilder = kinesisClientBuilder.endpointOverride(new
URI(kinesisEndpoint));
} catch (URISyntaxException e) {
- throw new IllegalArgumentException("URI syntax is not correctly
specified for endpoint: "
- + kinesisEndpoint, e);
+ throw new IllegalArgumentException("URI syntax is not correctly
specified for endpoint: " + kinesisEndpoint,
+ e);
}
}
_kinesisClient = kinesisClientBuilder.build();
+
+ int numRetries = Integer.parseInt(props.getProperty(NUM_RETRIES,
DEFAULT_NUM_RETRIES));
+ long retryDelayMs = Long.parseLong(props.getProperty(RETRY_DELAY_MILLIS,
DEFAULT_RETRY_DELAY_MILLIS));
+ _retryPolicy = new FixedDelayRetryPolicy(numRetries, retryDelayMs);
} catch (Exception e) {
_kinesisClient = null;
}
}
@Override
public void produce(String topic, byte[] payload) {
- PutRecordRequest putRecordRequest =
-
PutRecordRequest.builder().streamName(topic).data(SdkBytes.fromByteArray(payload))
- .partitionKey(UUID.randomUUID().toString()).build();
- PutRecordResponse putRecordResponse =
_kinesisClient.putRecord(putRecordRequest);
+ try {
+ _retryPolicy.attempt(() -> putRecord(topic, null, payload));
+ } catch (Exception e) {
+ LOGGER.error("Retries exhausted while pushing record in stream {}",
topic);
Review Comment:
the log line seems misleading. we don't know why the attempts to push record
into stream failed. is retry exhaustion the only reason to reach this point?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
