antgustech
commented
5 years ago Same thing here. 12.11, no errors are thrown but no logs are created either on my S3 bucket:
KinesisProducerConfiguration config = new KinesisProducerConfiguration()
.setRecordMaxBufferedTime(3000)
.setMaxConnections(5)
.setRequestTimeout(60000)
.setRegion("eu-west-1");
final KinesisProducer kinesis = new KinesisProducer(config);
System.out.println(kinesis.toString());
ByteBuffer data = ByteBuffer.wrap(("Test main java").getBytes());
kinesis.addUserRecord("myStream", "appStarts", data);
System.out.println("Done");But If I instead wait for callbacks like below, I ge the following errors:
KinesisProducerConfiguration config = new KinesisProducerConfiguration()
.setRecordMaxBufferedTime(3000)
.setMaxConnections(5)
.setRequestTimeout(60000)
.setRegion("eu-west-1");
KinesisProducer kinesis = new KinesisProducer(config);
Thread.sleep(2000);
FutureCallback<UserRecordResult> myCallback = new FutureCallback<UserRecordResult>() {
@Override
public void onFailure(Throwable t) {
System.out.println("Failed: " + t.toString());
System.out.println(t.getStackTrace().toString());
t.printStackTrace();
}
@Override
public void onSuccess(UserRecordResult result) {
System.out.println("Success: " + result.toString());
}
};
for (int i = 0; i < 10; ++i) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
ListenableFuture<UserRecordResult> f = kinesis.addUserRecord("myStream", "myPartitionKey", data);
// If the Future is complete by the time we call addCallback, the callback will be invoked immediately.
Futures.addCallback(f, myCallback);
}
for (int i = 0; i < 5; i++) {
try {
Thread.sleep(10000); //So I can wait and see the callbacks.
} catch (InterruptedException e) {
e.printStackTrace();
}
}Failed: com.amazonaws.services.kinesis.producer.UserRecordFailedException
[Ljava.lang.StackTraceElement;@71d70b5e
com.amazonaws.services.kinesis.producer.UserRecordFailedException
at com.amazonaws.services.kinesis.producer.KinesisProducer$MessageHandler.onPutRecordResult(KinesisProducer.java:197)
at com.amazonaws.services.kinesis.producer.KinesisProducer$MessageHandler.access$000(KinesisProducer.java:131)
at com.amazonaws.services.kinesis.producer.KinesisProducer$MessageHandler$1.run(KinesisProducer.java:138)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:834)I have tried older versions such as 10 but they cannot connect at all, I get connection errors.
cgpassante
MaximilianoFelice
ChengzhiZhao
Cory-Bradshaw
namedgraph
wikier
srinihacks
fmthoma
mkemaldurmus
Sometimes calling addUserRecord starts to throw:
The KPL does not seems to recover from this. All further calls to
addUserRecordalso fail. Restarting the KPL java process fixes the situation.This seems to happen when the kinesis stream is throttling requests so my guess is that the native process cant write to the stream quickly enough and runs out of memory. If that's the case my expectation would be that the native process should start to discard older data and of course that if the native process dies the KPL recovers to a working state.