@Test public void testRecordsSentToRecordWriterAndThenToProducer() throws IOException, SchemaNotFoundException, MalformedRecordException { final PublisherLease lease = new PublisherLease(producer, 1024 * 1024, 10L, logger, true, null, StandardCharsets.UTF_8); final FlowFile flowFile = new MockFlowFile(1L); final byte[] exampleInput = "101, John Doe, 48\n102, Jane Doe, 47".getBytes(StandardCharsets.UTF_8); final MockRecordParser readerService = new MockRecordParser(); readerService.addSchemaField("person_id", RecordFieldType.LONG); readerService.addSchemaField("name", RecordFieldType.STRING); readerService.addSchemaField("age", RecordFieldType.INT); final RecordReader reader = readerService.createRecordReader(Collections.emptyMap(), new ByteArrayInputStream(exampleInput), logger); final RecordSet recordSet = reader.createRecordSet(); final RecordSchema schema = reader.getSchema(); final String topic = "unit-test"; final String keyField = "person_id"; final RecordSetWriterFactory writerFactory = Mockito.mock(RecordSetWriterFactory.class); final RecordSetWriter writer = Mockito.mock(RecordSetWriter.class); Mockito.when(writer.write(Mockito.any(Record.class))).thenReturn(WriteResult.of(1, Collections.emptyMap())); Mockito.when(writerFactory.createWriter(eq(logger), eq(schema), any())).thenReturn(writer); lease.publish(flowFile, recordSet, writerFactory, schema, keyField, topic); verify(writerFactory, times(2)).createWriter(eq(logger), eq(schema), any()); verify(writer, times(2)).write(any(Record.class)); verify(producer, times(2)).send(any(), any()); }
void publish(final FlowFile flowFile, final InputStream flowFileContent, final byte[] messageKey, final byte[] demarcatorBytes, final String topic) throws IOException { if (tracker == null) { tracker = new InFlightMessageTracker(logger); } try { byte[] messageContent; if (demarcatorBytes == null || demarcatorBytes.length == 0) { if (flowFile.getSize() > maxMessageSize) { tracker.fail(flowFile, new TokenTooLargeException("A message in the stream exceeds the maximum allowed message size of " + maxMessageSize + " bytes.")); return; } messageContent = new byte[(int) flowFile.getSize()]; StreamUtils.fillBuffer(flowFileContent, messageContent); publish(flowFile, messageKey, messageContent, topic, tracker); return; } try (final StreamDemarcator demarcator = new StreamDemarcator(flowFileContent, demarcatorBytes, maxMessageSize)) { while ((messageContent = demarcator.nextToken()) != null) { publish(flowFile, messageKey, messageContent, topic, tracker); if (tracker.isFailed(flowFile)) { return; } } } catch (final TokenTooLargeException ttle) { tracker.fail(flowFile, ttle); } } catch (final Exception e) { tracker.fail(flowFile, e); poison(); throw e; } }
PublisherLease implements Closeable { void publish(final FlowFile flowFile, final InputStream flowFileContent, final byte[] messageKey, final byte[] demarcatorBytes, final String topic) throws IOException { if (tracker == null) { tracker = new InFlightMessageTracker(logger); } try { byte[] messageContent; if (demarcatorBytes == null || demarcatorBytes.length == 0) { if (flowFile.getSize() > maxMessageSize) { tracker.fail(flowFile, new TokenTooLargeException("A message in the stream exceeds the maximum allowed message size of " + maxMessageSize + " bytes.")); return; } messageContent = new byte[(int) flowFile.getSize()]; StreamUtils.fillBuffer(flowFileContent, messageContent); publish(flowFile, messageKey, messageContent, topic, tracker); return; } try (final StreamDemarcator demarcator = new StreamDemarcator(flowFileContent, demarcatorBytes, maxMessageSize)) { while ((messageContent = demarcator.nextToken()) != null) { publish(flowFile, messageKey, messageContent, topic, tracker); if (tracker.isFailed(flowFile)) { return; } } } catch (final TokenTooLargeException ttle) { tracker.fail(flowFile, ttle); } } catch (final Exception e) { tracker.fail(flowFile, e); poison(); throw e; } } }
PublisherLease implements Closeable { void publish(final FlowFile flowFile, final InputStream flowFileContent, final byte[] messageKey, final byte[] demarcatorBytes, final String topic) throws IOException { if (tracker == null) { tracker = new InFlightMessageTracker(logger); } try { byte[] messageContent; if (demarcatorBytes == null || demarcatorBytes.length == 0) { if (flowFile.getSize() > maxMessageSize) { tracker.fail(flowFile, new TokenTooLargeException("A message in the stream exceeds the maximum allowed message size of " + maxMessageSize + " bytes.")); return; } messageContent = new byte[(int) flowFile.getSize()]; StreamUtils.fillBuffer(flowFileContent, messageContent); publish(flowFile, messageKey, messageContent, topic, tracker); return; } try (final StreamDemarcator demarcator = new StreamDemarcator(flowFileContent, demarcatorBytes, maxMessageSize)) { while ((messageContent = demarcator.nextToken()) != null) { publish(flowFile, messageKey, messageContent, topic, tracker); if (tracker.isFailed(flowFile)) { return; } } } catch (final TokenTooLargeException ttle) { tracker.fail(flowFile, ttle); } } catch (final Exception e) { tracker.fail(flowFile, e); poison(); throw e; } } PublisherLease(final Producer<byte[], byte[]> producer, final int maxMessageSize, final long maxAckWaitMillis, final ComponentLog logger,
        final boolean useTransactions, final Pattern attributeNameRegex, final Charset headerCharacterSet); }
PublisherLease implements Closeable { void publish(final FlowFile flowFile, final InputStream flowFileContent, final byte[] messageKey, final byte[] demarcatorBytes, final String topic) throws IOException { if (tracker == null) { tracker = new InFlightMessageTracker(logger); } try { byte[] messageContent; if (demarcatorBytes == null || demarcatorBytes.length == 0) { if (flowFile.getSize() > maxMessageSize) { tracker.fail(flowFile, new TokenTooLargeException("A message in the stream exceeds the maximum allowed message size of " + maxMessageSize + " bytes.")); return; } messageContent = new byte[(int) flowFile.getSize()]; StreamUtils.fillBuffer(flowFileContent, messageContent); publish(flowFile, messageKey, messageContent, topic, tracker); return; } try (final StreamDemarcator demarcator = new StreamDemarcator(flowFileContent, demarcatorBytes, maxMessageSize)) { while ((messageContent = demarcator.nextToken()) != null) { publish(flowFile, messageKey, messageContent, topic, tracker); if (tracker.isFailed(flowFile)) { return; } } } catch (final TokenTooLargeException ttle) { tracker.fail(flowFile, ttle); } } catch (final Exception e) { tracker.fail(flowFile, e); poison(); throw e; } } PublisherLease(final Producer<byte[], byte[]> producer, final int maxMessageSize, final long maxAckWaitMillis, final ComponentLog logger,
        final boolean useTransactions, final Pattern attributeNameRegex, final Charset headerCharacterSet); boolean isPoisoned(); PublishResult complete(); @Override void close(); InFlightMessageTracker getTracker(); }
PublisherLease implements Closeable { void publish(final FlowFile flowFile, final InputStream flowFileContent, final byte[] messageKey, final byte[] demarcatorBytes, final String topic) throws IOException { if (tracker == null) { tracker = new InFlightMessageTracker(logger); } try { byte[] messageContent; if (demarcatorBytes == null || demarcatorBytes.length == 0) { if (flowFile.getSize() > maxMessageSize) { tracker.fail(flowFile, new TokenTooLargeException("A message in the stream exceeds the maximum allowed message size of " + maxMessageSize + " bytes.")); return; } messageContent = new byte[(int) flowFile.getSize()]; StreamUtils.fillBuffer(flowFileContent, messageContent); publish(flowFile, messageKey, messageContent, topic, tracker); return; } try (final StreamDemarcator demarcator = new StreamDemarcator(flowFileContent, demarcatorBytes, maxMessageSize)) { while ((messageContent = demarcator.nextToken()) != null) { publish(flowFile, messageKey, messageContent, topic, tracker); if (tracker.isFailed(flowFile)) { return; } } } catch (final TokenTooLargeException ttle) { tracker.fail(flowFile, ttle); } } catch (final Exception e) { tracker.fail(flowFile, e); poison(); throw e; } } PublisherLease(final Producer<byte[], byte[]> producer, final int maxMessageSize, final long maxAckWaitMillis, final ComponentLog logger,
        final boolean useTransactions, final Pattern attributeNameRegex, final Charset headerCharacterSet); boolean isPoisoned(); PublishResult complete(); @Override void close(); InFlightMessageTracker getTracker();  }