[ https://issues.apache.org/jira/browse/FLINK-5701?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15862844#comment-15862844 ]
ASF GitHub Bot commented on FLINK-5701: --------------------------------------- Github user tillrohrmann commented on a diff in the pull request: https://github.com/apache/flink/pull/3278#discussion_r100695754 --- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java --- @@ -88,195 +87,296 @@ public void testKeyValueDeserializersSetIfMissing() throws Exception { @Test public void testPartitionerOpenedWithDeterminatePartitionList() throws Exception { KafkaPartitioner mockPartitioner = mock(KafkaPartitioner.class); + RuntimeContext mockRuntimeContext = mock(RuntimeContext.class); when(mockRuntimeContext.getIndexOfThisSubtask()).thenReturn(0); when(mockRuntimeContext.getNumberOfParallelSubtasks()).thenReturn(1); + + // out-of-order list of 4 partitions + List<PartitionInfo> mockPartitionsList = new ArrayList<>(4); + mockPartitionsList.add(new PartitionInfo(DummyFlinkKafkaProducer.DUMMY_TOPIC, 3, null, null, null)); + mockPartitionsList.add(new PartitionInfo(DummyFlinkKafkaProducer.DUMMY_TOPIC, 1, null, null, null)); + mockPartitionsList.add(new PartitionInfo(DummyFlinkKafkaProducer.DUMMY_TOPIC, 0, null, null, null)); + mockPartitionsList.add(new PartitionInfo(DummyFlinkKafkaProducer.DUMMY_TOPIC, 2, null, null, null)); + + KafkaProducer mockProducer = mock(KafkaProducer.class); + when(mockProducer.partitionsFor(anyString())).thenReturn(mockPartitionsList); + when(mockProducer.metrics()).thenReturn(null); DummyFlinkKafkaProducer producer = new DummyFlinkKafkaProducer( - FakeStandardProducerConfig.get(), mockPartitioner); + FakeStandardProducerConfig.get(), mockPartitioner, mockProducer); producer.setRuntimeContext(mockRuntimeContext); producer.open(new Configuration()); - // the internal mock KafkaProducer will return an out-of-order list of 4 partitions, - // which should be sorted before provided to the custom partitioner's open() method + // the out-of-order partitions list should be sorted before provided to the custom partitioner's open() method int[] correctPartitionList = {0, 1, 2, 3}; verify(mockPartitioner).open(0, 1, correctPartitionList); } /** - * Test ensuring that the producer is not dropping buffered records.; - * we set a timeout because the test will not finish if the logic is broken + * Test ensuring that if an invoke call happens right after an async exception is caught, it should be rethrown */ - @Test(timeout=5000) - public void testAtLeastOnceProducer() throws Throwable { - runAtLeastOnceTest(true); + @Test + public void testAsyncErrorRethrownOnInvoke() throws Throwable { + KafkaProducer mockProducer = mock(KafkaProducer.class); + final DummyFlinkKafkaProducer<String> producer = new DummyFlinkKafkaProducer<>( + FakeStandardProducerConfig.get(), null, mockProducer); + + OneInputStreamOperatorTestHarness<String, Object> testHarness = + new OneInputStreamOperatorTestHarness<>(new StreamSink(producer)); + + testHarness.open(); + + testHarness.processElement(new StreamRecord<>("msg-1")); + + // let the message request return an async exception + producer.getPendingCallbacks().get(0).onCompletion(null, new Exception("artificial async exception")); + + try { + testHarness.processElement(new StreamRecord<>("msg-2")); + } catch (Exception e) { + // the next invoke should rethrow the async exception + Assert.assertTrue(e.getCause().getMessage().contains("artificial async exception")); + return; + } + + Assert.fail(); } /** - * Ensures that the at least once producing test fails if the flushing is disabled + * Test ensuring that if a snapshot call happens right after an async exception is caught, it should be rethrown */ - @Test(expected = AssertionError.class, timeout=5000) - public void testAtLeastOnceProducerFailsIfFlushingDisabled() throws Throwable { - runAtLeastOnceTest(false); - } - - private void runAtLeastOnceTest(boolean flushOnCheckpoint) throws Throwable { - final AtomicBoolean snapshottingFinished = new AtomicBoolean(false); + @Test + public void testAsyncErrorRethrownOnCheckpoint() throws Throwable { + KafkaProducer mockProducer = mock(KafkaProducer.class); final DummyFlinkKafkaProducer<String> producer = new DummyFlinkKafkaProducer<>( - FakeStandardProducerConfig.get(), null, snapshottingFinished); - producer.setFlushOnCheckpoint(flushOnCheckpoint); + FakeStandardProducerConfig.get(), null, mockProducer); OneInputStreamOperatorTestHarness<String, Object> testHarness = - new OneInputStreamOperatorTestHarness<>(new StreamSink(producer)); + new OneInputStreamOperatorTestHarness<>(new StreamSink(producer)); testHarness.open(); - for (int i = 0; i < 100; i++) { - testHarness.processElement(new StreamRecord<>("msg-" + i)); + testHarness.processElement(new StreamRecord<>("msg-1")); + + // let the message request return an async exception + producer.getPendingCallbacks().get(0).onCompletion(null, new Exception("artificial async exception")); + + try { + testHarness.snapshot(123L, 123L); + } catch (Exception e) { + // the next invoke should rethrow the async exception + Assert.assertTrue(e.getCause().getMessage().contains("artificial async exception")); + return; } - // start a thread confirming all pending records - final Tuple1<Throwable> runnableError = new Tuple1<>(null); - final Thread threadA = Thread.currentThread(); + Assert.fail(); + } - Runnable confirmer = new Runnable() { + /** + * Test ensuring that if an async exception is caught for one of the flushed requests on checkpoint, + * it should be rethrown; we set a timeout because the test will not finish if the logic is broken. + * + * Note that this test does not test the snapshot method is blocked correctly when there are pending recorrds. + * The test for that is covered in testAtLeastOnceProducer. + */ + @Test(timeout=5000) + public void testAsyncErrorRethrownOnCheckpointAfterFlush() throws Throwable { + KafkaProducer mockProducer = mock(KafkaProducer.class); + final DummyFlinkKafkaProducer<String> producer = new DummyFlinkKafkaProducer<>( + FakeStandardProducerConfig.get(), null, mockProducer); + producer.setFlushOnCheckpoint(true); + + final 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")); + + verify(mockProducer, times(3)).send(any(ProducerRecord.class), any(Callback.class)); + + // only let the first callback succeed for now + producer.getPendingCallbacks().get(0).onCompletion(null, null); + + final Tuple1<Throwable> asyncError = new Tuple1<>(null); + Thread snapshotThread = new Thread(new Runnable() { @Override public void run() { try { - MockProducer mp = producer.getProducerInstance(); - List<Callback> pending = mp.getPending(); - - // we need to find out if the snapshot() method blocks forever - // this is not possible. If snapshot() is running, it will - // start removing elements from the pending list. - synchronized (threadA) { - threadA.wait(500L); - } - // we now check that no records have been confirmed yet - Assert.assertEquals(100, pending.size()); - Assert.assertFalse("Snapshot method returned before all records were confirmed", - snapshottingFinished.get()); - - // now confirm all checkpoints - for (Callback c: pending) { - c.onCompletion(null, null); - } - pending.clear(); - } catch(Throwable t) { - runnableError.f0 = t; + // this should block at first, since there are still two pending records that needs to be flushed + testHarness.snapshot(123L, 123L); + } catch (Exception e) { + asyncError.f0 = e; } } - }; - Thread threadB = new Thread(confirmer); - threadB.start(); + }); + snapshotThread.start(); - // this should block: - testHarness.snapshot(0, 0); + // let the 2nd message fail with an async exception + producer.getPendingCallbacks().get(1).onCompletion(null, new Exception("artificial async failure for 2nd message")); + producer.getPendingCallbacks().get(2).onCompletion(null, null); - synchronized (threadA) { - threadA.notifyAll(); // just in case, to let the test fail faster - } - Assert.assertEquals(0, producer.getProducerInstance().getPending().size()); - Deadline deadline = FiniteDuration.apply(5, "s").fromNow(); - while (deadline.hasTimeLeft() && threadB.isAlive()) { - threadB.join(500); + snapshotThread.join(); + + // the snapshot should have failed with the async exception + Assert.assertTrue(asyncError.f0 != null && asyncError.f0.getCause().getMessage().contains("artificial async failure for 2nd message")); + } + + /** + * Test ensuring that the producer is not dropping buffered records; + * we set a timeout because the test will not finish if the logic is broken + */ + @Test(timeout=10000) + public void testAtLeastOnceProducer() throws Throwable { + + KafkaProducer mockProducer = mock(KafkaProducer.class); + final DummyFlinkKafkaProducer<String> producer = new DummyFlinkKafkaProducer<>( + FakeStandardProducerConfig.get(), null, mockProducer); + producer.setFlushOnCheckpoint(true); + + final 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")); + + verify(mockProducer, times(3)).send(any(ProducerRecord.class), any(Callback.class)); + Assert.assertEquals(3, producer.getPendingSize()); + + // start a thread to perform checkpointing + final Tuple1<Throwable> runnableError = new Tuple1<>(null); + final OneShotLatch snapshotReturnedLatch = new OneShotLatch(); + + Thread snapshotThread = new Thread(new Runnable() { + @Override + public void run() { + try { + // this should block until all records are flushed + testHarness.snapshot(123L, 123L); + } catch (Throwable e) { + runnableError.f0 = e; + } finally { + snapshotReturnedLatch.trigger(); + } + } + }); + snapshotThread.start(); + + // being extra safe that the snapshot is correctly blocked + try { + snapshotReturnedLatch.await(3, TimeUnit.SECONDS); + } catch (TimeoutException expected) { + // } - Assert.assertFalse("Thread A is expected to be finished at this point. If not, the test is prone to fail", threadB.isAlive()); + Assert.assertTrue("Snapshot returned before all records were flushed", !snapshotReturnedLatch.isTriggered()); + + producer.getPendingCallbacks().get(0).onCompletion(null, null); + Assert.assertTrue("Snapshot returned before all records were flushed", !snapshotReturnedLatch.isTriggered()); + Assert.assertEquals(2, producer.getPendingSize()); + + producer.getPendingCallbacks().get(1).onCompletion(null, null); + Assert.assertTrue("Snapshot returned before all records were flushed", !snapshotReturnedLatch.isTriggered()); + Assert.assertEquals(1, producer.getPendingSize()); + + producer.getPendingCallbacks().get(2).onCompletion(null, null); + Assert.assertEquals(0, producer.getPendingSize()); + + snapshotReturnedLatch.await(); + snapshotThread.join(); + if (runnableError.f0 != null) { throw runnableError.f0; } testHarness.close(); } + /** + * This test is meant to assure that testAtLeastOnceProducer is valid by testing that if flushing is disabled, + * the snapshot method does indeed finishes without waiting for pending records; + * we set a timeout because the test will not finish if the logic is broken + */ + @Test(timeout=5000) + public void testDoesNotWaitForPendingRecordsIfFlushingDisabled() throws Throwable { - // ------------------------------------------------------------------------ + KafkaProducer mockProducer = mock(KafkaProducer.class); + final DummyFlinkKafkaProducer<String> producer = new DummyFlinkKafkaProducer<>( + FakeStandardProducerConfig.get(), null, mockProducer); + producer.setFlushOnCheckpoint(false); - private static class DummyFlinkKafkaProducer<T> extends FlinkKafkaProducerBase<T> { - private static final long serialVersionUID = 1L; + final OneInputStreamOperatorTestHarness<String, Object> testHarness = + new OneInputStreamOperatorTestHarness<>(new StreamSink(producer)); - private transient MockProducer prod; - private AtomicBoolean snapshottingFinished; + testHarness.open(); - @SuppressWarnings("unchecked") - public DummyFlinkKafkaProducer(Properties producerConfig, KafkaPartitioner partitioner, AtomicBoolean snapshottingFinished) { - super("dummy-topic", (KeyedSerializationSchema< T >) mock(KeyedSerializationSchema.class), producerConfig, partitioner); - this.snapshottingFinished = snapshottingFinished; - } + testHarness.processElement(new StreamRecord<>("msg")); - // constructor variant for test irrelated to snapshotting - @SuppressWarnings("unchecked") - public DummyFlinkKafkaProducer(Properties producerConfig, KafkaPartitioner partitioner) { - super("dummy-topic", (KeyedSerializationSchema< T >) mock(KeyedSerializationSchema.class), producerConfig, partitioner); - this.snapshottingFinished = new AtomicBoolean(true); - } + // make sure that all callbacks have not been completed + verify(mockProducer, times(1)).send(any(ProducerRecord.class), any(Callback.class)); - @Override - protected <K, V> KafkaProducer<K, V> getKafkaProducer(Properties props) { - this.prod = new MockProducer(); - return this.prod; - } + // should return even if there are pending records + testHarness.snapshot(123L, 123L); - @Override - public void snapshotState(FunctionSnapshotContext ctx) throws Exception { - // call the actual snapshot state - super.snapshotState(ctx); - // notify test that snapshotting has been done - snapshottingFinished.set(true); - } + testHarness.close(); + } - @Override - protected void flush() { - this.prod.flush(); - } + // ------------------------------------------------------------------------ - public MockProducer getProducerInstance() { - return this.prod; - } - } + private static class DummyFlinkKafkaProducer<T> extends FlinkKafkaProducerBase<T> { + private static final long serialVersionUID = 1L; + + private final static String DUMMY_TOPIC = "dummy-topic"; - private static class MockProducer<K, V> extends KafkaProducer<K, V> { - List<Callback> pendingCallbacks = new ArrayList<>(); + private final KafkaProducer mockProducer; + private final List<Callback> pendingCallbacks = new ArrayList<>(); --- End diff -- `Callback` is not serializable. > FlinkKafkaProducer should check asyncException on checkpoints > ------------------------------------------------------------- > > Key: FLINK-5701 > URL: https://issues.apache.org/jira/browse/FLINK-5701 > Project: Flink > Issue Type: Bug > Components: Kafka Connector, Streaming Connectors > Reporter: Tzu-Li (Gordon) Tai > Priority: Critical > > Reported in ML: > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Fink-KafkaProducer-Data-Loss-td11413.html > The problem: > The producer holds a {{pendingRecords}} value that is incremented on each > invoke() and decremented on each callback, used to check if the producer > needs to sync on pending callbacks on checkpoints. > On each checkpoint, we should only consider the checkpoint succeeded iff > after flushing the {{pendingRecords == 0}} and {{asyncException == null}} > (currently, we’re only checking {{pendingRecords}}). > A quick fix for this is to check and rethrow async exceptions in the > {{snapshotState}} method both before and after flushing and > {{pendingRecords}} becomes 0. -- This message was sent by Atlassian JIRA (v6.3.15#6346)