NIFI-3739: This closes #1695. Added ConsumeKafkaRecord_0_10 and PublishKafkaRecord_0_10 processors
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/07989b84 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/07989b84 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/07989b84 Branch: refs/heads/master Commit: 07989b84601a41ea50038d456da3c0964d45ff83 Parents: 946f4a1 Author: Mark Payne <[email protected]> Authored: Tue Apr 25 17:19:41 2017 -0400 Committer: joewitt <[email protected]> Committed: Mon May 1 18:47:51 2017 -0400 ---------------------------------------------------------------------- .../language/StandardPropertyValue.java | 5 +- .../nifi/serialization/record/MapRecord.java | 24 +- .../nifi/serialization/record/Record.java | 3 + .../serialization/record/SerializedForm.java | 120 ++++++ .../record/util/DataTypeUtils.java | 107 ++++- .../java/org/apache/nifi/avro/AvroTypeUtil.java | 4 +- ...rtonworksEncodedSchemaReferenceStrategy.java | 2 +- .../nifi/serialization/DateTimeUtils.java | 19 +- .../nifi/controller/ProcessScheduler.java | 3 +- .../service/ControllerServiceNode.java | 5 +- .../service/ControllerServiceProvider.java | 7 +- .../apache/nifi/controller/FlowController.java | 5 +- .../nifi/controller/StandardProcessorNode.java | 9 +- .../scheduling/StandardProcessScheduler.java | 5 +- .../service/StandardControllerServiceNode.java | 10 +- .../StandardControllerServiceProvider.java | 53 ++- .../nifi-kafka-0-10-processors/pom.xml | 8 + .../kafka/pubsub/ConsumeKafkaRecord_0_10.java | 346 +++++++++++++++++ .../kafka/pubsub/ConsumeKafka_0_10.java | 8 +- .../processors/kafka/pubsub/ConsumerLease.java | 165 +++++++- .../processors/kafka/pubsub/ConsumerPool.java | 91 ++++- .../kafka/pubsub/PublishKafkaRecord_0_10.java | 386 +++++++++++++++++++ .../processors/kafka/pubsub/PublisherLease.java | 55 ++- .../org.apache.nifi.processor.Processor | 4 +- .../kafka/pubsub/ConsumeKafkaTest.java | 33 +- .../kafka/pubsub/ConsumerPoolTest.java | 33 +- .../pubsub/TestConsumeKafkaRecord_0_10.java | 219 +++++++++++ .../pubsub/TestPublishKafkaRecord_0_10.java | 287 ++++++++++++++ .../kafka/pubsub/util/MockRecordParser.java | 105 +++++ .../kafka/pubsub/util/MockRecordWriter.java | 103 +++++ .../services/AvroSchemaRegistry.java | 26 +- .../services/TestAvroSchemaRegistry.java | 44 --- .../hortonworks/HortonworksSchemaRegistry.java | 92 +++-- .../java/org/apache/nifi/avro/AvroReader.java | 56 ++- .../nifi/avro/AvroReaderWithExplicitSchema.java | 4 +- .../apache/nifi/avro/AvroRecordSetWriter.java | 74 +++- .../avro/WriteAvroResultWithExternalSchema.java | 19 + .../nifi/avro/WriteAvroResultWithSchema.java | 19 + .../org/apache/nifi/csv/CSVRecordReader.java | 13 +- .../org/apache/nifi/csv/CSVRecordSetWriter.java | 3 +- .../org/apache/nifi/csv/WriteCSVResult.java | 12 +- .../nifi/json/AbstractJsonRowRecordReader.java | 6 +- .../nifi/json/JsonPathRowRecordReader.java | 13 +- .../apache/nifi/json/JsonRecordSetWriter.java | 3 +- .../nifi/json/JsonTreeRowRecordReader.java | 54 ++- .../org/apache/nifi/json/WriteJsonResult.java | 64 ++- .../DateTimeTextRecordSetWriter.java | 19 +- .../SchemaRegistryRecordSetWriter.java | 4 +- .../nifi/json/TestJsonTreeRowRecordReader.java | 63 +++ .../apache/nifi/json/TestWriteJsonResult.java | 72 ++++ 50 files changed, 2579 insertions(+), 305 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPropertyValue.java ---------------------------------------------------------------------- diff --git a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPropertyValue.java b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPropertyValue.java index ac370bd..f6b6c70 100644 --- a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPropertyValue.java +++ b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPropertyValue.java @@ -149,13 +149,16 @@ public class StandardPropertyValue implements PropertyValue { } @Override + @SuppressWarnings("unchecked") public PropertyValue evaluateAttributeExpressions(FlowFile flowFile, Map<String, String> additionalAttributes, AttributeValueDecorator decorator, Map<String, String> stateValues) throws ProcessException { if (rawValue == null || preparedQuery == null) { return this; } + final ValueLookup lookup = new ValueLookup(variableRegistry, flowFile, additionalAttributes); - return new StandardPropertyValue(preparedQuery.evaluateExpressions(lookup, decorator, stateValues), serviceLookup, null); + final String evaluated = preparedQuery.evaluateExpressions(lookup, decorator, stateValues); + return new StandardPropertyValue(evaluated, serviceLookup, new EmptyPreparedQuery(evaluated), null); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/MapRecord.java ---------------------------------------------------------------------- diff --git a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/MapRecord.java b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/MapRecord.java index a6b965d..f9a22fc 100644 --- a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/MapRecord.java +++ b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/MapRecord.java @@ -27,10 +27,18 @@ import java.util.Optional; public class MapRecord implements Record { private final RecordSchema schema; private final Map<String, Object> values; + private final Optional<SerializedForm> serializedForm; public MapRecord(final RecordSchema schema, final Map<String, Object> values) { this.schema = Objects.requireNonNull(schema); this.values = Objects.requireNonNull(values); + this.serializedForm = Optional.empty(); + } + + public MapRecord(final RecordSchema schema, final Map<String, Object> values, final SerializedForm serializedForm) { + this.schema = Objects.requireNonNull(schema); + this.values = Objects.requireNonNull(values); + this.serializedForm = Optional.ofNullable(serializedForm); } @Override @@ -144,19 +152,12 @@ public class MapRecord implements Record { return convertToString(getValue(field), format); } - private String getFormat(final String optionalFormat, final RecordFieldType fieldType) { - return (optionalFormat == null) ? fieldType.getDefaultFormat() : optionalFormat; - } - private String convertToString(final Object value, final String format) { if (value == null) { return null; } - final String dateFormat = getFormat(format, RecordFieldType.DATE); - final String timestampFormat = getFormat(format, RecordFieldType.TIMESTAMP); - final String timeFormat = getFormat(format, RecordFieldType.TIME); - return DataTypeUtils.toString(value, dateFormat, timeFormat, timestampFormat); + return DataTypeUtils.toString(value, format); } @Override @@ -191,7 +192,7 @@ public class MapRecord implements Record { @Override public Date getAsDate(final String fieldName, final String format) { - return DataTypeUtils.toDate(getValue(fieldName), format, fieldName); + return DataTypeUtils.toDate(getValue(fieldName), DataTypeUtils.getDateFormat(format), fieldName); } @Override @@ -224,4 +225,9 @@ public class MapRecord implements Record { public String toString() { return "MapRecord[values=" + values + "]"; } + + @Override + public Optional<SerializedForm> getSerializedForm() { + return serializedForm; + } } http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/Record.java ---------------------------------------------------------------------- diff --git a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/Record.java b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/Record.java index 5e5e7ba..31aaab7 100644 --- a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/Record.java +++ b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/Record.java @@ -18,6 +18,7 @@ package org.apache.nifi.serialization.record; import java.util.Date; +import java.util.Optional; public interface Record { @@ -61,4 +62,6 @@ public interface Record { Date getAsDate(String fieldName, String format); Object[] getAsArray(String fieldName); + + Optional<SerializedForm> getSerializedForm(); } http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/SerializedForm.java ---------------------------------------------------------------------- diff --git a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/SerializedForm.java b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/SerializedForm.java new file mode 100644 index 0000000..438c895 --- /dev/null +++ b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/SerializedForm.java @@ -0,0 +1,120 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.serialization.record; + +import java.util.Objects; + +public interface SerializedForm { + /** + * @return the serialized form of the record. This could be a byte[], String, ByteBuffer, etc. + */ + Object getSerialized(); + + /** + * @return the MIME type that the data is serialized in + */ + String getMimeType(); + + public static SerializedForm of(final java.util.function.Supplier<Object> serializedSupplier, final String mimeType) { + Objects.requireNonNull(serializedSupplier); + Objects.requireNonNull(mimeType); + + return new SerializedForm() { + private volatile Object serialized = null; + + @Override + public Object getSerialized() { + if (serialized != null) { + return serialized; + } + + final Object supplied = serializedSupplier.get(); + this.serialized = supplied; + return supplied; + } + + @Override + public String getMimeType() { + return mimeType; + } + + @Override + public int hashCode() { + return 31 + 17 * mimeType.hashCode() + 15 * getSerialized().hashCode(); + } + + @Override + public boolean equals(final Object obj) { + if (obj == this) { + return true; + } + + if (obj == null) { + return false; + } + + if (!(obj instanceof SerializedForm)) { + return false; + } + + final SerializedForm other = (SerializedForm) obj; + return other.getMimeType().equals(mimeType) && Objects.deepEquals(other.getSerialized(), getSerialized()); + } + }; + } + + public static SerializedForm of(final Object serialized, final String mimeType) { + Objects.requireNonNull(serialized); + Objects.requireNonNull(mimeType); + + return new SerializedForm() { + @Override + public Object getSerialized() { + return serialized; + } + + @Override + public String getMimeType() { + return mimeType; + } + + @Override + public int hashCode() { + return 31 + 17 * mimeType.hashCode() + 15 * serialized.hashCode(); + } + + @Override + public boolean equals(final Object obj) { + if (obj == this) { + return true; + } + + if (obj == null) { + return false; + } + + if (!(obj instanceof SerializedForm)) { + return false; + } + + final SerializedForm other = (SerializedForm) obj; + return other.getMimeType().equals(mimeType) && Objects.deepEquals(other.getSerialized(), getSerialized()); + } + }; + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java ---------------------------------------------------------------------- diff --git a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java index a81e843..f59ac0d 100644 --- a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java +++ b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java @@ -43,10 +43,25 @@ public class DataTypeUtils { private static final TimeZone gmt = TimeZone.getTimeZone("gmt"); public static Object convertType(final Object value, final DataType dataType, final String fieldName) { - return convertType(value, dataType, RecordFieldType.DATE.getDefaultFormat(), RecordFieldType.TIME.getDefaultFormat(), RecordFieldType.TIMESTAMP.getDefaultFormat(), fieldName); + return convertType(value, dataType, getDateFormat(RecordFieldType.DATE.getDefaultFormat()), getDateFormat(RecordFieldType.TIME.getDefaultFormat()), + getDateFormat(RecordFieldType.TIMESTAMP.getDefaultFormat()), fieldName); } - public static Object convertType(final Object value, final DataType dataType, final String dateFormat, final String timeFormat, final String timestampFormat, final String fieldName) { + public static DateFormat getDateFormat(final RecordFieldType fieldType, final DateFormat dateFormat, final DateFormat timeFormat, final DateFormat timestampFormat) { + switch (fieldType) { + case DATE: + return dateFormat; + case TIME: + return timeFormat; + case TIMESTAMP: + return timestampFormat; + } + + return null; + } + + public static Object convertType(final Object value, final DataType dataType, final DateFormat dateFormat, final DateFormat timeFormat, + final DateFormat timestampFormat, final String fieldName) { switch (dataType.getFieldType()) { case BIGINT: return toBigInt(value, fieldName); @@ -69,7 +84,7 @@ public class DataTypeUtils { case SHORT: return toShort(value, fieldName); case STRING: - return toString(value, dateFormat, timeFormat, timestampFormat); + return toString(value, getDateFormat(dataType.getFieldType(), dateFormat, timeFormat, timestampFormat)); case TIME: return toTime(value, timeFormat, fieldName); case TIMESTAMP: @@ -273,7 +288,36 @@ public class DataTypeUtils { } - public static String toString(final Object value, final String dateFormat, final String timeFormat, final String timestampFormat) { + public static String toString(final Object value, final DateFormat format) { + if (value == null) { + return null; + } + + if (value instanceof String) { + return (String) value; + } + + if (format == null && value instanceof java.util.Date) { + return String.valueOf(((java.util.Date) value).getTime()); + } + + if (value instanceof java.sql.Date) { + return format.format((java.util.Date) value); + } + if (value instanceof java.sql.Time) { + return format.format((java.util.Date) value); + } + if (value instanceof java.sql.Timestamp) { + return format.format((java.util.Date) value); + } + if (value instanceof java.util.Date) { + return format.format((java.util.Date) value); + } + + return value.toString(); + } + + public static String toString(final Object value, final String format) { if (value == null) { return null; } @@ -282,17 +326,21 @@ public class DataTypeUtils { return (String) value; } + if (format == null && value instanceof java.util.Date) { + return String.valueOf(((java.util.Date) value).getTime()); + } + if (value instanceof java.sql.Date) { - return getDateFormat(dateFormat).format((java.util.Date) value); + return getDateFormat(format).format((java.util.Date) value); } if (value instanceof java.sql.Time) { - return getDateFormat(timeFormat).format((java.util.Date) value); + return getDateFormat(format).format((java.util.Date) value); } if (value instanceof java.sql.Timestamp) { - return getDateFormat(timestampFormat).format((java.util.Date) value); + return getDateFormat(format).format((java.util.Date) value); } if (value instanceof java.util.Date) { - return getDateFormat(timestampFormat).format((java.util.Date) value); + return getDateFormat(format).format((java.util.Date) value); } return value.toString(); @@ -302,7 +350,7 @@ public class DataTypeUtils { return value != null; } - public static java.sql.Date toDate(final Object value, final String format, final String fieldName) { + public static java.sql.Date toDate(final Object value, final DateFormat format, final String fieldName) { if (value == null) { return null; } @@ -318,9 +366,18 @@ public class DataTypeUtils { if (value instanceof String) { try { - final java.util.Date utilDate = getDateFormat(format).parse((String) value); + final String string = ((String) value).trim(); + if (string.isEmpty()) { + return null; + } + + if (format == null) { + return new Date(Long.parseLong(string)); + } + + final java.util.Date utilDate = format.parse(string); return new Date(utilDate.getTime()); - } catch (final ParseException e) { + } catch (final ParseException | NumberFormatException e) { throw new IllegalTypeConversionException("Could not convert value [" + value + "] of type java.lang.String to Date because the value is not in the expected date format: " + format + " for field " + fieldName); } @@ -350,7 +407,7 @@ public class DataTypeUtils { return false; } - public static Time toTime(final Object value, final String format, final String fieldName) { + public static Time toTime(final Object value, final DateFormat format, final String fieldName) { if (value == null) { return null; } @@ -366,7 +423,16 @@ public class DataTypeUtils { if (value instanceof String) { try { - final java.util.Date utilDate = getDateFormat(format).parse((String) value); + final String string = ((String) value).trim(); + if (string.isEmpty()) { + return null; + } + + if (format == null) { + return new Time(Long.parseLong(string)); + } + + final java.util.Date utilDate = format.parse(string); return new Time(utilDate.getTime()); } catch (final ParseException e) { throw new IllegalTypeConversionException("Could not convert value [" + value @@ -377,7 +443,7 @@ public class DataTypeUtils { throw new IllegalTypeConversionException("Cannot convert value [" + value + "] of type " + value.getClass() + " to Time for field " + fieldName); } - private static DateFormat getDateFormat(final String format) { + public static DateFormat getDateFormat(final String format) { final DateFormat df = new SimpleDateFormat(format); df.setTimeZone(gmt); return df; @@ -387,7 +453,7 @@ public class DataTypeUtils { return isDateTypeCompatible(value, format); } - public static Timestamp toTimestamp(final Object value, final String format, final String fieldName) { + public static Timestamp toTimestamp(final Object value, final DateFormat format, final String fieldName) { if (value == null) { return null; } @@ -403,7 +469,16 @@ public class DataTypeUtils { if (value instanceof String) { try { - final java.util.Date utilDate = getDateFormat(format).parse((String) value); + final String string = ((String) value).trim(); + if (string.isEmpty()) { + return null; + } + + if (format == null) { + return new Timestamp(Long.parseLong(string)); + } + + final java.util.Date utilDate = format.parse(string); return new Timestamp(utilDate.getTime()); } catch (final ParseException e) { throw new IllegalTypeConversionException("Could not convert value [" + value http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/main/java/org/apache/nifi/avro/AvroTypeUtil.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/main/java/org/apache/nifi/avro/AvroTypeUtil.java b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/main/java/org/apache/nifi/avro/AvroTypeUtil.java index 3af368e..bfdba3d 100644 --- a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/main/java/org/apache/nifi/avro/AvroTypeUtil.java +++ b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/main/java/org/apache/nifi/avro/AvroTypeUtil.java @@ -296,6 +296,8 @@ public class AvroTypeUtil { } return map; + } else if (rawValue instanceof Map) { + return rawValue; } else { throw new IllegalTypeConversionException("Cannot convert value " + rawValue + " of type " + rawValue.getClass() + " to a Map"); } @@ -358,7 +360,7 @@ public class AvroTypeUtil { case ENUM: return new GenericData.EnumSymbol(fieldSchema, rawValue); case STRING: - return DataTypeUtils.toString(rawValue, RecordFieldType.DATE.getDefaultFormat(), RecordFieldType.TIME.getDefaultFormat(), RecordFieldType.TIMESTAMP.getDefaultFormat()); + return DataTypeUtils.toString(rawValue, (String) null); } return rawValue; http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/schema/access/HortonworksEncodedSchemaReferenceStrategy.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/schema/access/HortonworksEncodedSchemaReferenceStrategy.java b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/schema/access/HortonworksEncodedSchemaReferenceStrategy.java index a00e322..de89900 100644 --- a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/schema/access/HortonworksEncodedSchemaReferenceStrategy.java +++ b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/schema/access/HortonworksEncodedSchemaReferenceStrategy.java @@ -60,7 +60,7 @@ public class HortonworksEncodedSchemaReferenceStrategy implements SchemaAccessSt final int protocolVersion = bb.get(); if (protocolVersion != 1) { throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. The latest known Protocol is Version " - + LATEST_PROTOCOL_VERSION + " but the data was encoded with version " + protocolVersion); + + LATEST_PROTOCOL_VERSION + " but the data was encoded with version " + protocolVersion + " or was not encoded with this data format"); } final long schemaId = bb.getLong(); http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/serialization/DateTimeUtils.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/serialization/DateTimeUtils.java b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/serialization/DateTimeUtils.java index d5ab8c5..efc3e4e 100644 --- a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/serialization/DateTimeUtils.java +++ b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/serialization/DateTimeUtils.java @@ -18,33 +18,32 @@ package org.apache.nifi.serialization; import org.apache.nifi.components.PropertyDescriptor; -import org.apache.nifi.serialization.record.RecordFieldType; public class DateTimeUtils { public static final PropertyDescriptor DATE_FORMAT = new PropertyDescriptor.Builder() .name("Date Format") - .description("Specifies the format to use when reading/writing Date fields") + .description("Specifies the format to use when reading/writing Date fields. " + + "If not specified, Date fields will be assumed to be number of milliseconds since epoch (Midnight, Jan 1, 1970 GMT).") .expressionLanguageSupported(false) - .defaultValue(RecordFieldType.DATE.getDefaultFormat()) .addValidator(new SimpleDateFormatValidator()) - .required(true) + .required(false) .build(); public static final PropertyDescriptor TIME_FORMAT = new PropertyDescriptor.Builder() .name("Time Format") - .description("Specifies the format to use when reading/writing Time fields") + .description("Specifies the format to use when reading/writing Time fields. " + + "If not specified, Time fields will be assumed to be number of milliseconds since epoch (Midnight, Jan 1, 1970 GMT).") .expressionLanguageSupported(false) - .defaultValue(RecordFieldType.TIME.getDefaultFormat()) .addValidator(new SimpleDateFormatValidator()) - .required(true) + .required(false) .build(); public static final PropertyDescriptor TIMESTAMP_FORMAT = new PropertyDescriptor.Builder() .name("Timestamp Format") - .description("Specifies the format to use when reading/writing Timestamp fields") + .description("Specifies the format to use when reading/writing Timestamp fields. " + + "If not specified, Timestamp fields will be assumed to be number of milliseconds since epoch (Midnight, Jan 1, 1970 GMT).") .expressionLanguageSupported(false) - .defaultValue(RecordFieldType.TIMESTAMP.getDefaultFormat()) .addValidator(new SimpleDateFormatValidator()) - .required(true) + .required(false) .build(); } http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ProcessScheduler.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ProcessScheduler.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ProcessScheduler.java index 6d98e46..5bb8981 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ProcessScheduler.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ProcessScheduler.java @@ -17,6 +17,7 @@ package org.apache.nifi.controller; import java.util.List; +import java.util.concurrent.CompletableFuture; import org.apache.nifi.connectable.Connectable; import org.apache.nifi.connectable.Funnel; @@ -162,7 +163,7 @@ public interface ProcessScheduler { * * @param service to enable */ - void enableControllerService(ControllerServiceNode service); + CompletableFuture<Void> enableControllerService(ControllerServiceNode service); /** * Disables all of the given Controller Services in the order provided by the List http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java index 8fe4eb2..faf530f 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java @@ -23,6 +23,7 @@ import org.apache.nifi.groups.ProcessGroup; import java.util.List; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ScheduledExecutorService; public interface ControllerServiceNode extends ConfiguredComponent { @@ -94,8 +95,10 @@ public interface ControllerServiceNode extends ConfiguredComponent { * initiate service enabling task as well as its re-tries * @param administrativeYieldMillis * the amount of milliseconds to wait for administrative yield + * + * @return a CompletableFuture that can be used to wait for the service to finish enabling */ - void enable(ScheduledExecutorService scheduler, long administrativeYieldMillis); + CompletableFuture<Void> enable(ScheduledExecutorService scheduler, long administrativeYieldMillis); /** * Will disable this service. Disabling of the service typically means http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java index 4169281..f7ba5e5 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java @@ -19,6 +19,7 @@ package org.apache.nifi.controller.service; import java.net.URL; import java.util.Collection; import java.util.Set; +import java.util.concurrent.Future; import org.apache.nifi.annotation.lifecycle.OnAdded; import org.apache.nifi.bundle.BundleCoordinate; @@ -65,11 +66,13 @@ public interface ControllerServiceProvider extends ControllerServiceLookup { /** * Enables the given controller service that it can be used by other - * components + * components. This method will asynchronously enable the service, returning + * immediately. * * @param serviceNode the service node + * @return a Future that can be used to wait for the service to finish being enabled. */ - void enableControllerService(ControllerServiceNode serviceNode); + Future<Void> enableControllerService(ControllerServiceNode serviceNode); /** * Enables the collection of services. If a service in this collection http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java index b628668..7853591 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java @@ -236,6 +236,7 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -3235,8 +3236,8 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R } @Override - public void enableControllerService(final ControllerServiceNode serviceNode) { - controllerServiceProvider.enableControllerService(serviceNode); + public Future<Void> enableControllerService(final ControllerServiceNode serviceNode) { + return controllerServiceProvider.enableControllerService(serviceNode); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java index 281c695..f42321e 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java @@ -125,6 +125,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable private final AtomicLong schedulingNanos; private final ProcessScheduler processScheduler; private long runNanos = 0L; + private volatile long yieldNanos; private final NiFiProperties nifiProperties; private SchedulingStrategy schedulingStrategy; // guarded by read/write lock @@ -518,7 +519,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable @Override public long getYieldPeriod(final TimeUnit timeUnit) { - return FormatUtils.getTimeDuration(getYieldPeriod(), timeUnit == null ? DEFAULT_TIME_UNIT : timeUnit); + final TimeUnit unit = (timeUnit == null ? DEFAULT_TIME_UNIT : timeUnit); + return unit.convert(yieldNanos, TimeUnit.NANOSECONDS); } @Override @@ -531,11 +533,12 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable if (isRunning()) { throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); } - final long yieldMillis = FormatUtils.getTimeDuration(requireNonNull(yieldPeriod), TimeUnit.MILLISECONDS); - if (yieldMillis < 0) { + final long yieldNanos = FormatUtils.getTimeDuration(requireNonNull(yieldPeriod), TimeUnit.NANOSECONDS); + if (yieldNanos < 0) { throw new IllegalArgumentException("Yield duration must be positive"); } this.yieldPeriod.set(yieldPeriod); + this.yieldNanos = yieldNanos; } /** http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java index 3cafbfe..5368d37 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java @@ -21,6 +21,7 @@ import static java.util.Objects.requireNonNull; import java.lang.reflect.InvocationTargetException; import java.util.List; import java.util.concurrent.Callable; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Future; @@ -535,8 +536,8 @@ public final class StandardProcessScheduler implements ProcessScheduler { } @Override - public void enableControllerService(final ControllerServiceNode service) { - service.enable(this.componentLifeCycleThreadPool, this.administrativeYieldMillis); + public CompletableFuture<Void> enableControllerService(final ControllerServiceNode service) { + return service.enable(this.componentLifeCycleThreadPool, this.administrativeYieldMillis); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java index 4ee65ec..7a744b7 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java @@ -54,6 +54,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map.Entry; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -382,7 +383,9 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i * as it reached ENABLED state. */ @Override - public void enable(final ScheduledExecutorService scheduler, final long administrativeYieldMillis) { + public CompletableFuture<Void> enable(final ScheduledExecutorService scheduler, final long administrativeYieldMillis) { + final CompletableFuture<Void> future = new CompletableFuture<>(); + if (this.stateRef.compareAndSet(ControllerServiceState.DISABLED, ControllerServiceState.ENABLING)) { synchronized (active) { this.active.set(true); @@ -396,6 +399,9 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i try (final NarCloseable nc = NarCloseable.withComponentNarLoader(getControllerServiceImplementation().getClass(), getIdentifier())) { ReflectionUtils.invokeMethodsWithAnnotation(OnEnabled.class, getControllerServiceImplementation(), configContext); } + + future.complete(null); + boolean shouldEnable = false; synchronized (active) { shouldEnable = active.get() && stateRef.compareAndSet(ControllerServiceState.ENABLING, ControllerServiceState.ENABLED); @@ -426,6 +432,8 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i } }); } + + return future; } /** http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java index ab70e75..4c6c3a3 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java @@ -16,6 +16,21 @@ */ package org.apache.nifi.controller.service; +import static java.util.Objects.requireNonNull; + +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.net.URL; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Future; + import org.apache.commons.lang3.ClassUtils; import org.apache.commons.lang3.StringUtils; import org.apache.nifi.annotation.lifecycle.OnAdded; @@ -51,20 +66,6 @@ import org.apache.nifi.util.ReflectionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.lang.reflect.Method; -import java.lang.reflect.Proxy; -import java.net.URL; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.HashSet; -import java.util.Iterator; -import java.util.List; -import java.util.Map; -import java.util.Set; - -import static java.util.Objects.requireNonNull; - public class StandardControllerServiceProvider implements ControllerServiceProvider { private static final Logger logger = LoggerFactory.getLogger(StandardControllerServiceProvider.class); @@ -322,9 +323,9 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi } @Override - public void enableControllerService(final ControllerServiceNode serviceNode) { + public Future<Void> enableControllerService(final ControllerServiceNode serviceNode) { serviceNode.verifyCanEnable(); - processScheduler.enableControllerService(serviceNode); + return processScheduler.enableControllerService(serviceNode); } @Override @@ -349,7 +350,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi this.enableControllerServiceDependenciesFirst(controllerServiceNode); } } catch (Exception e) { - logger.error("Failed to enable " + controllerServiceNode + " due to " + e); + logger.error("Failed to enable " + controllerServiceNode, e); if (this.bulletinRepo != null) { this.bulletinRepo.addBulletin(BulletinFactory.createBulletin("Controller Service", Severity.ERROR.name(), "Could not start " + controllerServiceNode + " due to " + e)); @@ -359,16 +360,28 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi } } - private void enableControllerServiceDependenciesFirst(ControllerServiceNode serviceNode) { + private Future<Void> enableControllerServiceDependenciesFirst(ControllerServiceNode serviceNode) { + final List<Future<Void>> futures = new ArrayList<>(); + for (ControllerServiceNode depNode : serviceNode.getRequiredControllerServices()) { if (!depNode.isActive()) { - this.enableControllerServiceDependenciesFirst(depNode); + futures.add(this.enableControllerServiceDependenciesFirst(depNode)); } } + if (logger.isDebugEnabled()) { logger.debug("Enabling " + serviceNode); } - this.enableControllerService(serviceNode); + + for (final Future<Void> future : futures) { + try { + future.get(); + } catch (final Exception e) { + // Nothing we can really do. Will attempt to enable this service anyway. + } + } + + return this.enableControllerService(serviceNode); } static List<List<ControllerServiceNode>> determineEnablingOrder(final Map<String, ControllerServiceNode> serviceNodeMap) { http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/pom.xml ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/pom.xml b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/pom.xml index 0cc1dd4..1a649f0 100644 --- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/pom.xml +++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/pom.xml @@ -28,6 +28,14 @@ </dependency> <dependency> <groupId>org.apache.nifi</groupId> + <artifactId>nifi-record-serialization-service-api</artifactId> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-record</artifactId> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> <artifactId>nifi-processor-utils</artifactId> </dependency> <dependency> http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaRecord_0_10.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaRecord_0_10.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaRecord_0_10.java new file mode 100644 index 0000000..0882870 --- /dev/null +++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaRecord_0_10.java @@ -0,0 +1,346 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.processors.kafka.pubsub; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.errors.WakeupException; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.nifi.annotation.behavior.DynamicProperty; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.WritesAttribute; +import org.apache.nifi.annotation.behavior.WritesAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.annotation.lifecycle.OnUnscheduled; +import org.apache.nifi.components.AllowableValue; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.serialization.RecordReaderFactory; +import org.apache.nifi.serialization.RecordSetWriterFactory; + +@CapabilityDescription("Consumes messages from Apache Kafka specifically built against the Kafka 0.10.x Consumer API. " + + "The complementary NiFi processor for sending messages is PublishKafka_0_10. Please note that, at this time, the Processor assumes that " + + "all records that are retrieved from a given partition have the same schema. If any of the Kafka messages are pulled but cannot be parsed or written with the " + + "configured Record Reader or Record Writer, the contents of the message will be written to a separate FlowFile, and that FlowFile will be transferred to the " + + "'parse.failure' relationship. Otherwise, each FlowFile is sent to the 'success' relationship and may contain many individual messages within the single FlowFile. " + + "A 'record.count' attribute is added to indicate how many messages are contained in the FlowFile.") +@Tags({"Kafka", "Get", "Record", "csv", "avro", "json", "Ingest", "Ingress", "Topic", "PubSub", "Consume", "0.10.x"}) +@WritesAttributes({ + @WritesAttribute(attribute = "record.count", description = "The number of records received"), + @WritesAttribute(attribute = "mime.type", description = "The MIME Type that is provided by the configured Record Writer"), + @WritesAttribute(attribute = KafkaProcessorUtils.KAFKA_PARTITION, description = "The partition of the topic the records are from"), + @WritesAttribute(attribute = KafkaProcessorUtils.KAFKA_TOPIC, description = "The topic records are from") +}) +@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN) +@DynamicProperty(name = "The name of a Kafka configuration property.", value = "The value of a given Kafka configuration property.", + description = "These properties will be added on the Kafka configuration after loading any provided configuration properties." + + " In the event a dynamic property represents a property that was already set, its value will be ignored and WARN message logged." + + " For the list of available Kafka properties please refer to: http://kafka.apache.org/documentation.html#configuration. ") +@SeeAlso({ConsumeKafka_0_10.class, PublishKafka_0_10.class, PublishKafkaRecord_0_10.class}) +public class ConsumeKafkaRecord_0_10 extends AbstractProcessor { + + static final AllowableValue OFFSET_EARLIEST = new AllowableValue("earliest", "earliest", "Automatically reset the offset to the earliest offset"); + static final AllowableValue OFFSET_LATEST = new AllowableValue("latest", "latest", "Automatically reset the offset to the latest offset"); + static final AllowableValue OFFSET_NONE = new AllowableValue("none", "none", "Throw exception to the consumer if no previous offset is found for the consumer's group"); + static final AllowableValue TOPIC_NAME = new AllowableValue("names", "names", "Topic is a full topic name or comma separated list of names"); + static final AllowableValue TOPIC_PATTERN = new AllowableValue("pattern", "pattern", "Topic is a regex using the Java Pattern syntax"); + + static final PropertyDescriptor TOPICS = new PropertyDescriptor.Builder() + .name("topic") + .displayName("Topic Name(s)") + .description("The name of the Kafka Topic(s) to pull from. More than one can be supplied if comma separated.") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(true) + .build(); + + static final PropertyDescriptor TOPIC_TYPE = new PropertyDescriptor.Builder() + .name("topic_type") + .displayName("Topic Name Format") + .description("Specifies whether the Topic(s) provided are a comma separated list of names or a single regular expression") + .required(true) + .allowableValues(TOPIC_NAME, TOPIC_PATTERN) + .defaultValue(TOPIC_NAME.getValue()) + .build(); + + static final PropertyDescriptor RECORD_READER = new PropertyDescriptor.Builder() + .name("record-reader") + .displayName("Record Reader") + .description("The Record Reader to use for incoming FlowFiles") + .identifiesControllerService(RecordReaderFactory.class) + .expressionLanguageSupported(false) + .required(true) + .build(); + + static final PropertyDescriptor RECORD_WRITER = new PropertyDescriptor.Builder() + .name("record-writer") + .displayName("Record Writer") + .description("The Record Writer to use in order to serialize the data before sending to Kafka") + .identifiesControllerService(RecordSetWriterFactory.class) + .expressionLanguageSupported(false) + .required(true) + .build(); + + static final PropertyDescriptor GROUP_ID = new PropertyDescriptor.Builder() + .name("group.id") + .displayName("Group ID") + .description("A Group ID is used to identify consumers that are within the same consumer group. Corresponds to Kafka's 'group.id' property.") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(false) + .build(); + + static final PropertyDescriptor AUTO_OFFSET_RESET = new PropertyDescriptor.Builder() + .name("auto.offset.reset") + .displayName("Offset Reset") + .description("Allows you to manage the condition when there is no initial offset in Kafka or if the current offset does not exist any " + + "more on the server (e.g. because that data has been deleted). Corresponds to Kafka's 'auto.offset.reset' property.") + .required(true) + .allowableValues(OFFSET_EARLIEST, OFFSET_LATEST, OFFSET_NONE) + .defaultValue(OFFSET_LATEST.getValue()) + .build(); + + static final PropertyDescriptor MAX_POLL_RECORDS = new PropertyDescriptor.Builder() + .name("max.poll.records") + .displayName("Max Poll Records") + .description("Specifies the maximum number of records Kafka should return in a single poll.") + .required(false) + .defaultValue("10000") + .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR) + .build(); + + static final PropertyDescriptor MAX_UNCOMMITTED_TIME = new PropertyDescriptor.Builder() + .name("max-uncommit-offset-wait") + .displayName("Max Uncommitted Time") + .description("Specifies the maximum amount of time allowed to pass before offsets must be committed. " + + "This value impacts how often offsets will be committed. Committing offsets less often increases " + + "throughput but also increases the window of potential data duplication in the event of a rebalance " + + "or JVM restart between commits. This value is also related to maximum poll records and the use " + + "of a message demarcator. When using a message demarcator we can have far more uncommitted messages " + + "than when we're not as there is much less for us to keep track of in memory.") + .required(false) + .defaultValue("1 secs") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + static final Relationship REL_SUCCESS = new Relationship.Builder() + .name("success") + .description("FlowFiles received from Kafka. Depending on demarcation strategy it is a flow file per message or a bundle of messages grouped by topic and partition.") + .build(); + static final Relationship REL_PARSE_FAILURE = new Relationship.Builder() + .name("parse.failure") + .description("If a message from Kafka cannot be parsed using the configured Record Reader, the contents of the " + + "message will be routed to this Relationship as its own individual FlowFile.") + .build(); + + static final List<PropertyDescriptor> DESCRIPTORS; + static final Set<Relationship> RELATIONSHIPS; + + private volatile ConsumerPool consumerPool = null; + private final Set<ConsumerLease> activeLeases = Collections.synchronizedSet(new HashSet<>()); + + static { + List<PropertyDescriptor> descriptors = new ArrayList<>(); + descriptors.add(KafkaProcessorUtils.BOOTSTRAP_SERVERS); + descriptors.add(TOPICS); + descriptors.add(TOPIC_TYPE); + descriptors.add(RECORD_READER); + descriptors.add(RECORD_WRITER); + descriptors.add(KafkaProcessorUtils.SECURITY_PROTOCOL); + descriptors.add(KafkaProcessorUtils.KERBEROS_PRINCIPLE); + descriptors.add(KafkaProcessorUtils.USER_PRINCIPAL); + descriptors.add(KafkaProcessorUtils.USER_KEYTAB); + descriptors.add(KafkaProcessorUtils.SSL_CONTEXT_SERVICE); + descriptors.add(GROUP_ID); + descriptors.add(AUTO_OFFSET_RESET); + descriptors.add(MAX_POLL_RECORDS); + descriptors.add(MAX_UNCOMMITTED_TIME); + DESCRIPTORS = Collections.unmodifiableList(descriptors); + + final Set<Relationship> rels = new HashSet<>(); + rels.add(REL_SUCCESS); + rels.add(REL_PARSE_FAILURE); + RELATIONSHIPS = Collections.unmodifiableSet(rels); + } + + @Override + public Set<Relationship> getRelationships() { + return RELATIONSHIPS; + } + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return DESCRIPTORS; + } + + @OnStopped + public void close() { + final ConsumerPool pool = consumerPool; + consumerPool = null; + if (pool != null) { + pool.close(); + } + } + + @Override + protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) { + return new PropertyDescriptor.Builder() + .description("Specifies the value for '" + propertyDescriptorName + "' Kafka Configuration.") + .name(propertyDescriptorName).addValidator(new KafkaProcessorUtils.KafkaConfigValidator(ConsumerConfig.class)).dynamic(true) + .build(); + } + + @Override + protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) { + return KafkaProcessorUtils.validateCommonProperties(validationContext); + } + + private synchronized ConsumerPool getConsumerPool(final ProcessContext context) { + ConsumerPool pool = consumerPool; + if (pool != null) { + return pool; + } + + return consumerPool = createConsumerPool(context, getLogger()); + } + + protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) { + final int maxLeases = context.getMaxConcurrentTasks(); + final long maxUncommittedTime = context.getProperty(MAX_UNCOMMITTED_TIME).asTimePeriod(TimeUnit.MILLISECONDS); + + final Map<String, Object> props = new HashMap<>(); + KafkaProcessorUtils.buildCommonKafkaProperties(context, ConsumerConfig.class, props); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName()); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName()); + final String topicListing = context.getProperty(ConsumeKafkaRecord_0_10.TOPICS).evaluateAttributeExpressions().getValue(); + final String topicType = context.getProperty(ConsumeKafkaRecord_0_10.TOPIC_TYPE).evaluateAttributeExpressions().getValue(); + final List<String> topics = new ArrayList<>(); + final String securityProtocol = context.getProperty(KafkaProcessorUtils.SECURITY_PROTOCOL).getValue(); + final String bootstrapServers = context.getProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS).getValue(); + + final RecordReaderFactory readerFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class); + final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class); + + if (topicType.equals(TOPIC_NAME.getValue())) { + for (final String topic : topicListing.split(",", 100)) { + final String trimmedName = topic.trim(); + if (!trimmedName.isEmpty()) { + topics.add(trimmedName); + } + } + + return new ConsumerPool(maxLeases, readerFactory, writerFactory, props, topics, maxUncommittedTime, securityProtocol, bootstrapServers, log); + } else if (topicType.equals(TOPIC_PATTERN.getValue())) { + final Pattern topicPattern = Pattern.compile(topicListing.trim()); + return new ConsumerPool(maxLeases, readerFactory, writerFactory, props, topicPattern, maxUncommittedTime, securityProtocol, bootstrapServers, log); + } else { + getLogger().error("Subscription type has an unknown value {}", new Object[] {topicType}); + return null; + } + } + + @OnUnscheduled + public void interruptActiveThreads() { + // There are known issues with the Kafka client library that result in the client code hanging + // indefinitely when unable to communicate with the broker. In order to address this, we will wait + // up to 30 seconds for the Threads to finish and then will call Consumer.wakeup() to trigger the + // thread to wakeup when it is blocked, waiting on a response. + final long nanosToWait = TimeUnit.SECONDS.toNanos(5L); + final long start = System.nanoTime(); + while (System.nanoTime() - start < nanosToWait && !activeLeases.isEmpty()) { + try { + Thread.sleep(100L); + } catch (final InterruptedException ie) { + Thread.currentThread().interrupt(); + return; + } + } + + if (!activeLeases.isEmpty()) { + int count = 0; + for (final ConsumerLease lease : activeLeases) { + getLogger().info("Consumer {} has not finished after waiting 30 seconds; will attempt to wake-up the lease", new Object[] {lease}); + lease.wakeup(); + count++; + } + + getLogger().info("Woke up {} consumers", new Object[] {count}); + } + + activeLeases.clear(); + } + + @Override + public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { + final ConsumerPool pool = getConsumerPool(context); + if (pool == null) { + context.yield(); + return; + } + + try (final ConsumerLease lease = pool.obtainConsumer(session, context)) { + if (lease == null) { + context.yield(); + return; + } + + activeLeases.add(lease); + try { + while (this.isScheduled() && lease.continuePolling()) { + lease.poll(); + } + if (this.isScheduled() && !lease.commit()) { + context.yield(); + } + } catch (final WakeupException we) { + getLogger().warn("Was interrupted while trying to communicate with Kafka with lease {}. " + + "Will roll back session and discard any partially received data.", new Object[] {lease}); + } catch (final KafkaException kex) { + getLogger().error("Exception while interacting with Kafka so will close the lease {} due to {}", + new Object[]{lease, kex}, kex); + } catch (final Throwable t) { + getLogger().error("Exception while processing data from kafka so will close the lease {} due to {}", + new Object[]{lease, t}, t); + } finally { + activeLeases.remove(lease); + } + } + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka_0_10.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka_0_10.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka_0_10.java index 4da485f..f678fa3 100644 --- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka_0_10.java +++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka_0_10.java @@ -166,9 +166,9 @@ public class ConsumeKafka_0_10 extends AbstractProcessor { .build(); static final Relationship REL_SUCCESS = new Relationship.Builder() - .name("success") - .description("FlowFiles received from Kafka. Depending on demarcation strategy it is a flow file per message or a bundle of messages grouped by topic and partition.") - .build(); + .name("success") + .description("FlowFiles received from Kafka. Depending on demarcation strategy it is a flow file per message or a bundle of messages grouped by topic and partition.") + .build(); static final List<PropertyDescriptor> DESCRIPTORS; static final Set<Relationship> RELATIONSHIPS; @@ -305,7 +305,7 @@ public class ConsumeKafka_0_10 extends AbstractProcessor { return; } - try (final ConsumerLease lease = pool.obtainConsumer(session)) { + try (final ConsumerLease lease = pool.obtainConsumer(session, context)) { if (lease == null) { context.yield(); return; http://git-wip-us.apache.org/repos/asf/nifi/blob/07989b84/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java index 7ea180d..b2665ae 100644 --- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java +++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java @@ -16,15 +16,30 @@ */ package org.apache.nifi.processors.kafka.pubsub; +import static org.apache.nifi.processors.kafka.pubsub.ConsumeKafkaRecord_0_10.REL_PARSE_FAILURE; +import static org.apache.nifi.processors.kafka.pubsub.ConsumeKafkaRecord_0_10.REL_SUCCESS; +import static org.apache.nifi.processors.kafka.pubsub.KafkaProcessorUtils.HEX_ENCODING; +import static org.apache.nifi.processors.kafka.pubsub.KafkaProcessorUtils.UTF8_ENCODING; + +import java.io.BufferedOutputStream; +import java.io.ByteArrayInputStream; import java.io.Closeable; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + import javax.xml.bind.DatatypeConverter; + import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -34,11 +49,19 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.TopicPartition; import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.flowfile.attributes.CoreAttributes; import org.apache.nifi.logging.ComponentLog; import org.apache.nifi.processor.ProcessSession; -import static org.apache.nifi.processors.kafka.pubsub.ConsumeKafka_0_10.REL_SUCCESS; -import static org.apache.nifi.processors.kafka.pubsub.KafkaProcessorUtils.HEX_ENCODING; -import static org.apache.nifi.processors.kafka.pubsub.KafkaProcessorUtils.UTF8_ENCODING; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.serialization.RecordReader; +import org.apache.nifi.serialization.RecordReaderFactory; +import org.apache.nifi.serialization.RecordSetWriter; +import org.apache.nifi.serialization.RecordSetWriterFactory; +import org.apache.nifi.serialization.SimpleRecordSchema; +import org.apache.nifi.serialization.WriteResult; +import org.apache.nifi.serialization.record.Record; +import org.apache.nifi.serialization.record.RecordSchema; +import org.apache.nifi.serialization.record.RecordSet; /** * This class represents a lease to access a Kafka Consumer object. The lease is @@ -56,6 +79,8 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe private final String keyEncoding; private final String securityProtocol; private final String bootstrapServers; + private final RecordSetWriterFactory writerFactory; + private final RecordReaderFactory readerFactory; private boolean poisoned = false; //used for tracking demarcated flowfiles to their TopicPartition so we can append //to them on subsequent poll calls @@ -72,6 +97,8 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe final String keyEncoding, final String securityProtocol, final String bootstrapServers, + final RecordReaderFactory readerFactory, + final RecordSetWriterFactory writerFactory, final ComponentLog logger) { this.maxWaitMillis = maxWaitMillis; this.kafkaConsumer = kafkaConsumer; @@ -79,6 +106,8 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe this.keyEncoding = keyEncoding; this.securityProtocol = securityProtocol; this.bootstrapServers = bootstrapServers; + this.readerFactory = readerFactory; + this.writerFactory = writerFactory; this.logger = logger; } @@ -175,7 +204,9 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe getProcessSession().transfer(bundledFlowFiles, REL_SUCCESS); } getProcessSession().commit(); - kafkaConsumer.commitSync(uncommittedOffsetsMap); + + final Map<TopicPartition, OffsetAndMetadata> offsetsMap = uncommittedOffsetsMap; + kafkaConsumer.commitSync(offsetsMap); resetInternalState(); return true; } catch (final KafkaException kex) { @@ -269,8 +300,9 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe public abstract ProcessSession getProcessSession(); - private void processRecords(final ConsumerRecords<byte[], byte[]> records) { + public abstract void yield(); + private void processRecords(final ConsumerRecords<byte[], byte[]> records) { records.partitions().stream().forEach(partition -> { List<ConsumerRecord<byte[], byte[]>> messages = records.records(partition); if (!messages.isEmpty()) { @@ -279,17 +311,20 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe .mapToLong(record -> record.offset()) .max() .getAsLong(); - uncommittedOffsetsMap.put(partition, new OffsetAndMetadata(maxOffset + 1L)); //write records to content repository and session - if (demarcatorBytes == null) { + if (demarcatorBytes != null) { + writeDemarcatedData(getProcessSession(), messages, partition); + } else if (readerFactory != null && writerFactory != null) { + writeRecordData(getProcessSession(), messages, partition); + } else { totalFlowFiles += messages.size(); messages.stream().forEach(message -> { writeData(getProcessSession(), message, partition); }); - } else { - writeData(getProcessSession(), messages, partition); } + + uncommittedOffsetsMap.put(partition, new OffsetAndMetadata(maxOffset + 1L)); } }); } @@ -329,7 +364,7 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe session.transfer(tracker.flowFile, REL_SUCCESS); } - private void writeData(final ProcessSession session, final List<ConsumerRecord<byte[], byte[]>> records, final TopicPartition topicPartition) { + private void writeDemarcatedData(final ProcessSession session, final List<ConsumerRecord<byte[], byte[]>> records, final TopicPartition topicPartition) { final ConsumerRecord<byte[], byte[]> firstRecord = records.get(0); final boolean demarcateFirstRecord; BundleTracker tracker = bundleMap.get(topicPartition); @@ -343,6 +378,7 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe demarcateFirstRecord = true; //have already been writing records for this topic/partition in this lease } flowFile = tracker.flowFile; + tracker.incrementRecordCount(records.size()); flowFile = session.append(flowFile, out -> { boolean useDemarcator = demarcateFirstRecord; @@ -358,6 +394,115 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe bundleMap.put(topicPartition, tracker); } + private void rollback(final TopicPartition topicPartition) { + OffsetAndMetadata offsetAndMetadata = uncommittedOffsetsMap.get(topicPartition); + if (offsetAndMetadata == null) { + offsetAndMetadata = kafkaConsumer.committed(topicPartition); + } + + final long offset = offsetAndMetadata.offset(); + kafkaConsumer.seek(topicPartition, offset); + } + + private void writeRecordData(final ProcessSession session, final List<ConsumerRecord<byte[], byte[]>> records, final TopicPartition topicPartition) { + FlowFile flowFile = session.create(); + try { + final RecordSetWriter writer; + try { + writer = writerFactory.createWriter(logger, flowFile, new ByteArrayInputStream(new byte[0])); + } catch (final Exception e) { + logger.error( + "Failed to obtain a Record Writer for serializing Kafka messages. This generally happens because the " + + "Record Writer cannot obtain the appropriate Schema, due to failure to connect to a remote Schema Registry " + + "or due to the Schema Access Strategy being dependent upon FlowFile Attributes that are not available. " + + "Will roll back the Kafka message offsets.", e); + + try { + rollback(topicPartition); + } catch (final Exception rollbackException) { + logger.warn("Attempted to rollback Kafka message offset but was unable to do so", rollbackException); + } + + yield(); + throw new ProcessException(e); + } + + final FlowFile ff = flowFile; + final AtomicReference<WriteResult> writeResult = new AtomicReference<>(); + + flowFile = session.write(flowFile, rawOut -> { + final Iterator<ConsumerRecord<byte[], byte[]>> itr = records.iterator(); + + 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 Exception e) { + final Map<String, String> attributes = new HashMap<>(); + attributes.put(KafkaProcessorUtils.KAFKA_OFFSET, String.valueOf(consumerRecord.offset())); + attributes.put(KafkaProcessorUtils.KAFKA_PARTITION, String.valueOf(topicPartition.partition())); + attributes.put(KafkaProcessorUtils.KAFKA_TOPIC, topicPartition.topic()); + + FlowFile failureFlowFile = session.create(); + failureFlowFile = session.write(failureFlowFile, out -> out.write(consumerRecord.value())); + failureFlowFile = session.putAllAttributes(failureFlowFile, attributes); + + session.transfer(failureFlowFile, REL_PARSE_FAILURE); + logger.error("Failed to parse message from Kafka using the configured Record Reader. " + + "Will route message as its own FlowFile to the 'parse.failure' relationship", e); + + session.adjustCounter("Parse Failures", 1, false); + } + } + + return null; + } + }; + + try (final OutputStream out = new BufferedOutputStream(rawOut)) { + writeResult.set(writer.write(recordSet, out)); + } + }); + + final WriteResult result = writeResult.get(); + if (result.getRecordCount() > 0) { + final Map<String, String> attributes = new HashMap<>(result.getAttributes()); + attributes.put(CoreAttributes.MIME_TYPE.key(), writer.getMimeType()); + attributes.put("record.count", String.valueOf(result.getRecordCount())); + + attributes.put(KafkaProcessorUtils.KAFKA_PARTITION, String.valueOf(topicPartition.partition())); + attributes.put(KafkaProcessorUtils.KAFKA_TOPIC, topicPartition.topic()); + + flowFile = session.putAllAttributes(flowFile, attributes); + + final long executionDurationMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - leaseStartNanos); + final String transitUri = KafkaProcessorUtils.buildTransitURI(securityProtocol, bootstrapServers, topicPartition.topic()); + session.getProvenanceReporter().receive(flowFile, transitUri, executionDurationMillis); + + session.adjustCounter("Records Received", result.getRecordCount(), false); + session.transfer(flowFile, REL_SUCCESS); + } else { + session.remove(flowFile); + } + } catch (final Exception e) { + session.remove(flowFile); + throw e; + } + } + private void populateAttributes(final BundleTracker tracker) { final Map<String, String> kafkaAttrs = new HashMap<>(); kafkaAttrs.put(KafkaProcessorUtils.KAFKA_OFFSET, String.valueOf(tracker.initialOffset));
