FlinkKinesisProducer extends RichSinkFunction<OUT> implements CheckpointedFunction { @Override public void open(Configuration parameters) throws Exception { super.open(parameters); KinesisProducerConfiguration producerConfig = KinesisConfigUtil.getValidatedProducerConfiguration(configProps); producer = getKinesisProducer(producerConfig); callback = new FutureCallback<UserRecordResult>() { @Override public void onSuccess(UserRecordResult result) { if (!result.isSuccessful()) { if (failOnError) { if (thrownException == null) { thrownException = new RuntimeException("Record was not sent successful"); } } else { LOG.warn("Record was not sent successful"); } } } @Override public void onFailure(Throwable t) { if (failOnError) { thrownException = t; } else { LOG.warn("An exception occurred while processing a record", t); } } }; if (this.customPartitioner != null) { this.customPartitioner.initialize(getRuntimeContext().getIndexOfThisSubtask(), getRuntimeContext().getNumberOfParallelSubtasks()); } LOG.info("Started Kinesis producer instance for region '{}'", producerConfig.getRegion()); } FlinkKinesisProducer(final SerializationSchema<OUT> schema, Properties configProps);  FlinkKinesisProducer(KinesisSerializationSchema<OUT> schema, Properties configProps); void setFailOnError(boolean failOnError); void setDefaultStream(String defaultStream); void setDefaultPartition(String defaultPartition); void setCustomPartitioner(KinesisPartitioner<OUT> partitioner); @Override void open(Configuration parameters); @Override void invoke(OUT value); @Override void close(); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext context);  }
@SuppressWarnings("ResultOfMethodCallIgnored") @Test public void testAsyncErrorRethrownOnInvoke() throws Throwable { final DummyFlinkKinesisProducer<String> producer = new DummyFlinkKinesisProducer<>(new SimpleStringSchema()); OneInputStreamOperatorTestHarness<String, Object> testHarness = new OneInputStreamOperatorTestHarness<>(new StreamSink<>(producer)); testHarness.open(); testHarness.processElement(new StreamRecord<>("msg-1")); producer.getPendingRecordFutures().get(0).setException(new Exception("artificial async exception")); try { testHarness.processElement(new StreamRecord<>("msg-2")); } catch (Exception e) { Assert.assertTrue(ExceptionUtils.findThrowableWithMessage(e, "artificial async exception").isPresent()); return; } Assert.fail(); }
@SuppressWarnings("ResultOfMethodCallIgnored") @Test public void testAsyncErrorRethrownOnCheckpoint() throws Throwable { final DummyFlinkKinesisProducer<String> producer = new DummyFlinkKinesisProducer<>(new SimpleStringSchema()); OneInputStreamOperatorTestHarness<String, Object> testHarness = new OneInputStreamOperatorTestHarness<>(new StreamSink<>(producer)); testHarness.open(); testHarness.processElement(new StreamRecord<>("msg-1")); producer.getPendingRecordFutures().get(0).setException(new Exception("artificial async exception")); try { testHarness.snapshot(123L, 123L); } catch (Exception e) { Assert.assertTrue(ExceptionUtils.findThrowableWithMessage(e, "artificial async exception").isPresent()); return; } Assert.fail(); }
@SuppressWarnings("ResultOfMethodCallIgnored") @Test(timeout = 10000) public void testAsyncErrorRethrownAfterFlush() throws Throwable { final DummyFlinkKinesisProducer<String> producer = new DummyFlinkKinesisProducer<>(new SimpleStringSchema()); OneInputStreamOperatorTestHarness<String, Object> testHarness = new OneInputStreamOperatorTestHarness<>(new StreamSink<>(producer)); testHarness.open(); testHarness.processElement(new StreamRecord<>("msg-1")); testHarness.processElement(new StreamRecord<>("msg-2")); testHarness.processElement(new StreamRecord<>("msg-3")); UserRecordResult result = mock(UserRecordResult.class); when(result.isSuccessful()).thenReturn(true); producer.getPendingRecordFutures().get(0).set(result); CheckedThread snapshotThread = new CheckedThread() { @Override public void go() throws Exception { testHarness.snapshot(123L, 123L); } }; snapshotThread.start(); producer.getPendingRecordFutures().get(1).setException(new Exception("artificial async failure for 2nd message")); producer.getPendingRecordFutures().get(2).set(mock(UserRecordResult.class)); try { snapshotThread.sync(); } catch (Exception e) { Assert.assertTrue(ExceptionUtils.findThrowableWithMessage(e, "artificial async failure for 2nd message").isPresent()); return; } Assert.fail(); }