@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(); }
@Override public void open(Configuration parameters) throws Exception { super.open(parameters); KinesisProducerConfiguration producerConfig = KinesisConfigUtil.getValidatedProducerConfiguration(configProps); producer = getKinesisProducer(producerConfig); final MetricGroup kinesisMectricGroup = getRuntimeContext().getMetricGroup().addGroup(KINESIS_PRODUCER_METRIC_GROUP); this.backpressureCycles = kinesisMectricGroup.counter(METRIC_BACKPRESSURE_CYCLES); kinesisMectricGroup.gauge(METRIC_OUTSTANDING_RECORDS_COUNT, producer::getOutstandingRecordsCount); backpressureLatch = new TimeoutLatch(); callback = new FutureCallback<UserRecordResult>() { @Override public void onSuccess(UserRecordResult result) { backpressureLatch.trigger(); 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) { backpressureLatch.trigger(); 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 extends RichSinkFunction<OUT> implements CheckpointedFunction { @Override public void open(Configuration parameters) throws Exception { super.open(parameters); KinesisProducerConfiguration producerConfig = KinesisConfigUtil.getValidatedProducerConfiguration(configProps); producer = getKinesisProducer(producerConfig); final MetricGroup kinesisMectricGroup = getRuntimeContext().getMetricGroup().addGroup(KINESIS_PRODUCER_METRIC_GROUP); this.backpressureCycles = kinesisMectricGroup.counter(METRIC_BACKPRESSURE_CYCLES); kinesisMectricGroup.gauge(METRIC_OUTSTANDING_RECORDS_COUNT, producer::getOutstandingRecordsCount); backpressureLatch = new TimeoutLatch(); callback = new FutureCallback<UserRecordResult>() { @Override public void onSuccess(UserRecordResult result) { backpressureLatch.trigger(); 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) { backpressureLatch.trigger(); 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 extends RichSinkFunction<OUT> implements CheckpointedFunction { @Override public void open(Configuration parameters) throws Exception { super.open(parameters); KinesisProducerConfiguration producerConfig = KinesisConfigUtil.getValidatedProducerConfiguration(configProps); producer = getKinesisProducer(producerConfig); final MetricGroup kinesisMectricGroup = getRuntimeContext().getMetricGroup().addGroup(KINESIS_PRODUCER_METRIC_GROUP); this.backpressureCycles = kinesisMectricGroup.counter(METRIC_BACKPRESSURE_CYCLES); kinesisMectricGroup.gauge(METRIC_OUTSTANDING_RECORDS_COUNT, producer::getOutstandingRecordsCount); backpressureLatch = new TimeoutLatch(); callback = new FutureCallback<UserRecordResult>() { @Override public void onSuccess(UserRecordResult result) { backpressureLatch.trigger(); 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) { backpressureLatch.trigger(); 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); }
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); final MetricGroup kinesisMectricGroup = getRuntimeContext().getMetricGroup().addGroup(KINESIS_PRODUCER_METRIC_GROUP); this.backpressureCycles = kinesisMectricGroup.counter(METRIC_BACKPRESSURE_CYCLES); kinesisMectricGroup.gauge(METRIC_OUTSTANDING_RECORDS_COUNT, producer::getOutstandingRecordsCount); backpressureLatch = new TimeoutLatch(); callback = new FutureCallback<UserRecordResult>() { @Override public void onSuccess(UserRecordResult result) { backpressureLatch.trigger(); 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) { backpressureLatch.trigger(); 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 setQueueLimit(int queueLimit); void setDefaultStream(String defaultStream); void setDefaultPartition(String defaultPartition); void setCustomPartitioner(KinesisPartitioner<OUT> partitioner); @Override void open(Configuration parameters); @Override void invoke(OUT value, Context context); @Override void close(); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext context); }
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); final MetricGroup kinesisMectricGroup = getRuntimeContext().getMetricGroup().addGroup(KINESIS_PRODUCER_METRIC_GROUP); this.backpressureCycles = kinesisMectricGroup.counter(METRIC_BACKPRESSURE_CYCLES); kinesisMectricGroup.gauge(METRIC_OUTSTANDING_RECORDS_COUNT, producer::getOutstandingRecordsCount); backpressureLatch = new TimeoutLatch(); callback = new FutureCallback<UserRecordResult>() { @Override public void onSuccess(UserRecordResult result) { backpressureLatch.trigger(); 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) { backpressureLatch.trigger(); 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 setQueueLimit(int queueLimit); void setDefaultStream(String defaultStream); void setDefaultPartition(String defaultPartition); void setCustomPartitioner(KinesisPartitioner<OUT> partitioner); @Override void open(Configuration parameters); @Override void invoke(OUT value, Context context); @Override void close(); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext context); static final String KINESIS_PRODUCER_METRIC_GROUP; static final String METRIC_BACKPRESSURE_CYCLES; static final String METRIC_OUTSTANDING_RECORDS_COUNT; }