@Test public void testItemFailureRethrownOnCheckpoint() throws Throwable { final DummyElasticsearchSink<String> sink = new DummyElasticsearchSink<>( new HashMap<String, String>(), new SimpleSinkFunction<String>(), new NoOpFailureHandler()); final OneInputStreamOperatorTestHarness<String, Object> testHarness = new OneInputStreamOperatorTestHarness<>(new StreamSink<>(sink)); testHarness.open(); sink.setMockItemFailuresListForNextBulkItemResponses(Collections.singletonList(new Exception("artificial failure for record"))); testHarness.processElement(new StreamRecord<>("msg")); verify(sink.getMockBulkProcessor(), times(1)).add(any(IndexRequest.class)); sink.manualBulkRequestWithAllPendingRequests(); try { testHarness.snapshot(1L, 1000L); } catch (Exception e) { Assert.assertTrue(e.getCause().getCause().getMessage().contains("artificial failure for record")); return; } Assert.fail(); }
@Override public void open(Configuration parameters) throws Exception { client = callBridge.createClient(userConfig); bulkProcessor = buildBulkProcessor(new BulkProcessorListener()); requestIndexer = callBridge.createBulkProcessorIndexer(bulkProcessor, flushOnCheckpoint, numPendingRequests); failureRequestIndexer = new BufferingNoOpRequestIndexer(); }
ElasticsearchSinkBase extends RichSinkFunction<T> implements CheckpointedFunction { @Override public void open(Configuration parameters) throws Exception { client = callBridge.createClient(userConfig); bulkProcessor = buildBulkProcessor(new BulkProcessorListener()); requestIndexer = callBridge.createBulkProcessorIndexer(bulkProcessor, flushOnCheckpoint, numPendingRequests); failureRequestIndexer = new BufferingNoOpRequestIndexer(); } }
ElasticsearchSinkBase extends RichSinkFunction<T> implements CheckpointedFunction { @Override public void open(Configuration parameters) throws Exception { client = callBridge.createClient(userConfig); bulkProcessor = buildBulkProcessor(new BulkProcessorListener()); requestIndexer = callBridge.createBulkProcessorIndexer(bulkProcessor, flushOnCheckpoint, numPendingRequests); failureRequestIndexer = new BufferingNoOpRequestIndexer(); } ElasticsearchSinkBase(
		ElasticsearchApiCallBridge<C> callBridge,
		Map<String, String> userConfig,
		ElasticsearchSinkFunction<T> elasticsearchSinkFunction,
		ActionRequestFailureHandler failureHandler); }
ElasticsearchSinkBase extends RichSinkFunction<T> implements CheckpointedFunction { @Override public void open(Configuration parameters) throws Exception { client = callBridge.createClient(userConfig); bulkProcessor = buildBulkProcessor(new BulkProcessorListener()); requestIndexer = callBridge.createBulkProcessorIndexer(bulkProcessor, flushOnCheckpoint, numPendingRequests); failureRequestIndexer = new BufferingNoOpRequestIndexer(); } ElasticsearchSinkBase(
		ElasticsearchApiCallBridge<C> callBridge,
		Map<String, String> userConfig,
		ElasticsearchSinkFunction<T> elasticsearchSinkFunction,
		ActionRequestFailureHandler failureHandler); void disableFlushOnCheckpoint(); @Override void open(Configuration parameters); @Override void invoke(T value, Context context); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext context); @Override void close(); }
ElasticsearchSinkBase extends RichSinkFunction<T> implements CheckpointedFunction { @Override public void open(Configuration parameters) throws Exception { client = callBridge.createClient(userConfig); bulkProcessor = buildBulkProcessor(new BulkProcessorListener()); requestIndexer = callBridge.createBulkProcessorIndexer(bulkProcessor, flushOnCheckpoint, numPendingRequests); failureRequestIndexer = new BufferingNoOpRequestIndexer(); } ElasticsearchSinkBase(
		ElasticsearchApiCallBridge<C> callBridge,
		Map<String, String> userConfig,
		ElasticsearchSinkFunction<T> elasticsearchSinkFunction,
		ActionRequestFailureHandler failureHandler); void disableFlushOnCheckpoint(); @Override void open(Configuration parameters); @Override void invoke(T value, Context context); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext context); @Override void close(); static final String CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS; static final String CONFIG_KEY_BULK_FLUSH_MAX_SIZE_MB; static final String CONFIG_KEY_BULK_FLUSH_INTERVAL_MS; static final String CONFIG_KEY_BULK_FLUSH_BACKOFF_ENABLE; static final String CONFIG_KEY_BULK_FLUSH_BACKOFF_TYPE; static final String CONFIG_KEY_BULK_FLUSH_BACKOFF_RETRIES; static final String CONFIG_KEY_BULK_FLUSH_BACKOFF_DELAY; }