Github user joewitt commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1695#discussion_r114151657
--- Diff:
nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java
---
@@ -408,53 +424,63 @@ private void writeRecordData(final ProcessSession
session, final List<ConsumerRe
flowFile = session.write(flowFile, rawOut -> {
final Iterator<ConsumerRecord<byte[], byte[]>> itr =
records.iterator();
- try (final OutputStream out = new
BufferedOutputStream(rawOut)) {
- final RecordSchema emptySchema = new
SimpleRecordSchema(Collections.emptyList());
- final RecordSet recordSet = new RecordSet() {
- @Override
- public RecordSchema getSchema() throws IOException
{
- return emptySchema;
- }
-
- @Override
- public Record next() throws IOException {
- if (!itr.hasNext()) {
- return null;
- }
+ final RecordSchema emptySchema = new
SimpleRecordSchema(Collections.emptyList());
+ final RecordSet recordSet = new RecordSet() {
+ @Override
+ public RecordSchema getSchema() throws IOException {
+ return emptySchema;
+ }
+ @Override
+ public Record next() throws IOException {
+ while (itr.hasNext()) {
final ConsumerRecord<byte[], byte[]>
consumerRecord = itr.next();
final InputStream in = new
ByteArrayInputStream(consumerRecord.value());
try {
final RecordReader reader =
readerFactory.createRecordReader(ff, in, logger);
final Record record = reader.nextRecord();
return record;
- } catch (final SchemaNotFoundException |
MalformedRecordException e) {
- throw new IOException(e);
+ } catch (final Exception e) {
+ FlowFile failureFlowFile =
session.create();
--- End diff --
lets add the attributes we know for this message (topic/partition/offset)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
