This is an automated email from the ASF dual-hosted git repository.
bbende pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push:
new 06e2776583 NIFI-11828 - Confluent Schema Encoding Access Strategy -
Schema ID versus Schema Version ID (#7495)
06e2776583 is described below
commit 06e2776583ce426a24e4a0d2b4da7520ae8457de
Author: Pierre Villard <[email protected]>
AuthorDate: Tue Jul 25 15:42:50 2023 +0200
NIFI-11828 - Confluent Schema Encoding Access Strategy - Schema ID versus
Schema Version ID (#7495)
* NIFI-11828 - Confluent Schema Encoding Access Strategy - Schema ID versus
Schema Version ID
* removed hard coded version 1
---
.../nifi/confluent/schemaregistry/ConfluentSchemaRegistry.java | 2 +-
.../apache/nifi/schema/access/ConfluentSchemaRegistryStrategy.java | 5 ++---
.../nifi/schema/access/TestConfluentSchemaRegistryStrategy.java | 5 ++---
3 files changed, 5 insertions(+), 7 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistry.java
b/nifi-nar-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistry.java
index 65d5b9dc56..43335418b5 100644
---
a/nifi-nar-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistry.java
+++
b/nifi-nar-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistry.java
@@ -296,7 +296,7 @@ public class ConfluentSchemaRegistry extends
AbstractControllerService implement
}
private RecordSchema retrieveSchemaById(final SchemaIdentifier
schemaIdentifier) throws IOException, SchemaNotFoundException {
- final OptionalLong schemaId = schemaIdentifier.getIdentifier();
+ final OptionalLong schemaId = schemaIdentifier.getSchemaVersionId();
if (!schemaId.isPresent()) {
throw new
org.apache.nifi.schema.access.SchemaNotFoundException("Cannot retrieve schema
because Schema Id is not present");
}
diff --git
a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/schema/access/ConfluentSchemaRegistryStrategy.java
b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/schema/access/ConfluentSchemaRegistryStrategy.java
index 2734103be9..1e694f42d1 100644
---
a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/schema/access/ConfluentSchemaRegistryStrategy.java
+++
b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/main/java/org/apache/nifi/schema/access/ConfluentSchemaRegistryStrategy.java
@@ -54,7 +54,7 @@ public class ConfluentSchemaRegistryStrategy implements
SchemaAccessStrategy {
// This encoding follows the pattern that is provided for serializing
data by the Confluent Schema Registry serializer
// as it is provided at:
- //
http://docs.confluent.io/current/schema-registry/docs/serializer-formatter.html#wire-format
+ //
https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format
// The format consists of the first byte always being 0, to indicate a
'magic byte' followed by 4 bytes
// representing the schema id.
final ByteBuffer bb = ByteBuffer.wrap(buffer);
@@ -67,8 +67,7 @@ public class ConfluentSchemaRegistryStrategy implements
SchemaAccessStrategy {
final int schemaId = bb.getInt();
final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder()
- .id(Long.valueOf(schemaId))
- .version(1)
+ .schemaVersionId(Long.valueOf(schemaId))
.build();
return schemaRegistry.retrieveSchema(schemaIdentifier);
diff --git
a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/test/java/org/apache/nifi/schema/access/TestConfluentSchemaRegistryStrategy.java
b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/test/java/org/apache/nifi/schema/access/TestConfluentSchemaRegistryStrategy.java
index 997b2964aa..8a48bab1fe 100644
---
a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/test/java/org/apache/nifi/schema/access/TestConfluentSchemaRegistryStrategy.java
+++
b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-standard-record-utils/src/test/java/org/apache/nifi/schema/access/TestConfluentSchemaRegistryStrategy.java
@@ -47,10 +47,9 @@ public class TestConfluentSchemaRegistryStrategy extends
AbstractSchemaAccessStr
try (final ByteArrayInputStream in = new
ByteArrayInputStream(bytesOut.toByteArray())) {
- // the confluent strategy will read the id from the input
stream and use '1' as the version
+ // the confluent strategy will read the id from the input
stream
final SchemaIdentifier expectedSchemaIdentifier =
SchemaIdentifier.builder()
- .id((long)schemaId)
- .version(1)
+ .schemaVersionId((long)schemaId)
.build();
when(schemaRegistry.retrieveSchema(argThat(new
SchemaIdentifierMatcher(expectedSchemaIdentifier))))
