Вот код Java: < /p>
Код: Выделить всё
package com.versa.apicurio.confluent;
import com.versa.apicurio.confluent.Employee;
import com.versa.apicurio.confluent.DepartmentEnum;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
public class VersionBasedKafkaProducer {
private static final String REGISTRY_URL = "https://apicurio-sr.vkp.versa-vani.com";
private static final String LETS_ENCRYPT_ROOT_URL = "https://letsencrypt.org/certs/isrgrootx1.pem";
private static final String LETS_ENCRYPT_R11_URL = "https://letsencrypt.org/certs/2024/r11.pem";
private static final String TRUSTSTORE_PASSWORD = "changeit";
private static final String TRUSTSTORE_PATH = "apicurio-truststore.jks";
/**
* Creates a custom truststore with Let's Encrypt certificates
* @return The path to the created truststore
*/
private static String createCustomTruststore() throws Exception {
}
/**
* Downloads an X.509 certificate from a URL
*/
private static X509Certificate downloadCertificate(String certUrl) throws Exception {
}
/**
* Get the Access Token for authentication
*/
public static String getAccessToken(String tokenUrl, String clientId, String clientSecret) throws Exception {
String params = "grant_type=client_credentials"
+ "&client_id=" + clientId
+ "&client_secret=" + clientSecret;
URL url = new URL(tokenUrl);
HttpURLConnection conn = (HttpURLConnection) url.openConnection();
conn.setRequestMethod("POST");
conn.setDoOutput(true);
conn.setRequestProperty("Content-Type", "application/x-www-form-urlencoded");
try (OutputStream os = conn.getOutputStream()) {
os.write(params.getBytes(StandardCharsets.UTF_8));
}
try (Scanner scanner = new Scanner(conn.getInputStream(), StandardCharsets.UTF_8.name())) {
String resp = scanner.useDelimiter("\\A").next();
return resp.split("\"access_token\":\"")[1].split("\"")[0];
}
}
apiv
/**
* Verify schema exists and get information about it, returning the schema ID
*/
private static int verifySchemaAndGetId(String token) throws Exception {
System.out.println("\n=== Verifying schema in registry ===");
// Subject name with RecordNameStrategy
String subject = "com.versa.apicurio.confluent.Employee";
// Check if the subject exists
String subjectsUrl = String.format("%s/apis/ccompat/v7/subjects", REGISTRY_URL);
System.out.println("Checking all subjects at: " + subjectsUrl);
URL url = new URL(subjectsUrl);
HttpURLConnection conn = (HttpURLConnection) url.openConnection();
conn.setRequestMethod("GET");
conn.setRequestProperty("Authorization", "Bearer " + token);
int responseCode = conn.getResponseCode();
System.out.println("Response Code: " + responseCode);
boolean subjectExists = false;
if (responseCode == 200) {
try (Scanner scanner = new Scanner(conn.getInputStream(), StandardCharsets.UTF_8.name())) {
String resp = scanner.useDelimiter("\\A").next();
System.out.println("All subjects: " + resp);
// Check if our subject is in the list
subjectExists = resp.contains("\"" + subject + "\"");
System.out.println("Subject '" + subject + "' exists: " + subjectExists);
}
} else {
System.out.println("Failed to retrieve subjects list. Response code: " + responseCode);
return -1;
}
if (!subjectExists) {
System.out.println("Subject '" + subject + "' does not exist in the registry!");
return -1;
}
// Get versions
String versionsUrl = String.format("%s/apis/ccompat/v7/subjects/%s/versions", REGISTRY_URL, subject);
System.out.println("Fetching versions at: " + versionsUrl);
URL versionsUrlObj = new URL(versionsUrl);
HttpURLConnection versionsConn = (HttpURLConnection) versionsUrlObj.openConnection();
versionsConn.setRequestMethod("GET");
versionsConn.setRequestProperty("Authorization", "Bearer " + token);
int versionsResponseCode = versionsConn.getResponseCode();
String highestVersion = "1";
if (versionsResponseCode == 200) {
try (Scanner scanner = new Scanner(versionsConn.getInputStream(), StandardCharsets.UTF_8.name())) {
String resp = scanner.useDelimiter("\\A").next();
System.out.println("Available versions: " + resp);
if (resp.equals("[]") || resp.trim().isEmpty()) {
System.out.println("Warning: No versions found for this subject!");
return -1;
} else {
// Find the highest version number
Pattern pattern = Pattern.compile("\\d+");
Matcher matcher = pattern.matcher(resp);
int maxVersion = 0;
while (matcher.find()) {
try {
int version = Integer.parseInt(matcher.group());
if (version > maxVersion) {
maxVersion = version;
highestVersion = matcher.group();
}
} catch (NumberFormatException e) {
// Skip if not a valid number
}
}
System.out.println("Highest version found: " + highestVersion);
}
}
} else {
System.out.println("Failed to list versions. Response code: " + versionsResponseCode);
return -1;
}
// Get the specific version
String versionUrl = String.format("%s/apis/ccompat/v7/subjects/%s/versions/%s",
REGISTRY_URL, subject, highestVersion);
System.out.println("Fetching schema at: " + versionUrl);
URL specificVersionUrl = new URL(versionUrl);
HttpURLConnection versionConn = (HttpURLConnection) specificVersionUrl.openConnection();
versionConn.setRequestMethod("GET");
versionConn.setRequestProperty("Authorization", "Bearer " + token);
int versionResponseCode = versionConn.getResponseCode();
int schemaId = -1;
if (versionResponseCode == 200) {
try (Scanner scanner = new Scanner(versionConn.getInputStream(), StandardCharsets.UTF_8.name())) {
String resp = scanner.useDelimiter("\\A").next();
System.out.println("Version " + highestVersion + " schema (raw): " + resp);
// Extract schema ID
if (resp.contains("\"id\":")) {
String idStr = resp.split("\"id\":")[1].split(",")[0].trim();
try {
schemaId = Integer.parseInt(idStr);
System.out.println("Schema ID: " + schemaId);
} catch (NumberFormatException e) {
System.out.println("Failed to parse schema ID: " + idStr);
return -1;
}
}
// Extract and pretty print the schema
try {
String schemaJson = resp.split("\"schema\":\"")[1].split("\",\"schemaType\"")[0];
// Replace escaped quotes with regular quotes
schemaJson = schemaJson.replace("\\\"", "\"");
// Replace escaped backslashes with regular backslashes
schemaJson = schemaJson.replace("\\\\", "\\");
System.out.println("\nSchema content (prettified):\n" + schemaJson);
} catch (Exception e) {
System.out.println("Failed to prettify schema: " + e.getMessage());
}
}
return schemaId;
} else {
System.out.println("Failed to get version. Response code: " + versionResponseCode);
return -1;
}
}
public static void main(String[] args) throws Exception {
// Create truststore for SSL
String truststorePath = createCustomTruststore();
// Get token
String tokenUrl = "https://keycloak.vkp.versa-vani.com/realms/readonly-realm/protocol/openid-connect/token";
String clientId = "apicurio-registry";
String clientSecret = "";
String token = getAccessToken(tokenUrl, clientId, clientSecret);
System.out.println("Got access token: " + token.substring(0, 15) + "...");
// Get schema ID during verification
int schemaId = verifySchemaAndGetId(token);
// System.out.println("schemaId", schemaId);
if (schemaId
Вот ошибка: < /p>
(base) Karans-MacBook-Pro:apicurio_confluent karanalang$
(base) Karans-MacBook-Pro:apicurio_confluent karanalang$ mvn exec:java -Dexec.mainClass="com.versa.apicurio.confluent.VersionBasedKafkaProducer" -Dlog4j.debug
[INFO] Scanning for projects...
[INFO]
[INFO] ----------< com.versa.apicurio.confluent:apicurio_confluent >-----------
[INFO] Building apicurio_confluent 1.0-SNAPSHOT
[INFO] --------------------------------[ jar ]---------------------------------
[INFO]
[INFO] --- exec-maven-plugin:3.5.0:java (default-cli) @ apicurio_confluent ---
Creating custom truststore with Let's Encrypt certificates...
Downloading Let's Encrypt Root certificate...
Downloading Let's Encrypt R11 certificate...
Custom truststore created at: /Users/karanalang/Documents/Technology/apicurio-schema-registry/SchemaRegistry/apicurio_confluent/apicurio-truststore.jks
Got access token: eyJhbGciOiJSUzI...
=== Verifying schema in registry ===
Checking all subjects at: https://apicurio-sr.vkp.versa-vani.com/apis/ccompat/v7/subjects
Response Code: 200
All subjects: ["com.versa.apicurio.confluent.Employee"]
Subject 'com.versa.apicurio.confluent.Employee' exists: true
Fetching versions at: https://apicurio-sr.vkp.versa-vani.com/apis/ccompat/v7/subjects/com.versa.apicurio.confluent.Employee/versions
Available versions: [1,2,3,4]
Highest version found: 4
Fetching schema at: https://apicurio-sr.vkp.versa-vani.com/apis/ccompat/v7/subjects/com.versa.apicurio.confluent.Employee/versions/4
Version 4 schema (raw): {"subject":"com.versa.apicurio.confluent.Employee","version":4,"id":2,"schema":"{\"type\": \"record\", \"name\": \"Employee\", \"namespace\": \"com.versa.apicurio.confluent\", \"fields\": [{\"name\": \"id\", \"type\": \"int\"}, {\"name\": \"name\", \"type\": \"string\"}, {\"name\": \"salary\", \"type\": [\"null\", \"float\"], \"default\": null}, {\"name\": \"age\", \"type\": [\"null\", \"int\"], \"default\": null}, {\"name\": \"department\", \"type\": {\"type\": \"enum\", \"name\": \"DepartmentEnum\", \"symbols\": [\"HR\", \"ENGINEERING\", \"SALES\"]}}, {\"name\": \"email\", \"type\": [\"null\", \"string\"], \"default\": null}, {\"name\": \"new_col\", \"type\": [\"null\", \"string\"], \"default\": null}, {\"name\": \"new_col2\", \"type\": [\"null\", \"string\"], \"default\": null}, {\"name\": \"new_col3\", \"type\": [\"null\", \"string\"], \"default\": null}]}","schemaType":"AVRO","references":[]}
Schema ID: 2
Schema content (prettified):
{"type": "record", "name": "Employee", "namespace": "com.versa.apicurio.confluent", "fields": [{"name": "id", "type": "int"}, {"name": "name", "type": "string"}, {"name": "salary", "type": ["null", "float"], "default": null}, {"name": "age", "type": ["null", "int"], "default": null}, {"name": "department", "type": {"type": "enum", "name": "DepartmentEnum", "symbols": ["HR", "ENGINEERING", "SALES"]}}, {"name": "email", "type": ["null", "string"], "default": null}, {"name": "new_col", "type": ["null", "string"], "default": null}, {"name": "new_col2", "type": ["null", "string"], "default": null}, {"name": "new_col3", "type": ["null", "string"], "default": null}]}
=== Testing direct schema retrieval by ID ===
Testing URL: https://apicurio-sr.vkp.versa-vani.com/apis/ccompat/v7/schemas/ids/2
Schema ID check response code: 200
Schema by ID response: {"schema":"{\"type\": \"record\", \"name\": \"Employee\", \"namespace\": \"com.versa.apicurio.confluent\", \"fields\": [{\"name\": \"id\", \"type\": \"int\"}, {\"name\": \"name\", \"type\": \"string\"}, {\"name\": \"salary\", \"type\": [\"null\", \"float\"], \"default\": null}, {\"name\": \"age\", \"type\": [\"null\", \"int\"], \"default\": null}, {\"name\": \"department\", \"type\": {\"type\": \"enum\", \"name\": \"DepartmentEnum\", \"symbols\": [\"HR\", \"ENGINEERING\", \"SALES\"]}}, {\"name\": \"email\", \"type\": [\"null\", \"string\"], \"default\": null}, {\"name\": \"new_col\", \"type\": [\"null\", \"string\"], \"default\": null}, {\"name\": \"new_col2\", \"type\": [\"null\", \"string\"], \"default\": null}, {\"name\": \"new_col3\", \"type\": [\"null\", \"string\"], \"default\": null}]}","schemaType":"AVRO","references":[]}
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/Users/karanalang/.m2/repository/org/slf4j/slf4j-reload4j/1.7.36/slf4j-reload4j-1.7.36.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/Users/karanalang/.m2/repository/ch/qos/logback/logback-classic/1.2.10/logback-classic-1.2.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
log4j: Trying to find [log4j.xml] using context classloader org.codehaus.mojo.exec.URLClassLoaderBuilder$ExecJavaClassLoader@34fe326d.
log4j: Trying to find [log4j.xml] using org.codehaus.mojo.exec.URLClassLoaderBuilder$ExecJavaClassLoader@34fe326d class loader.
log4j: Trying to find [log4j.xml] using ClassLoader.getSystemResource().
log4j: Trying to find [log4j.properties] using context classloader org.codehaus.mojo.exec.URLClassLoaderBuilder$ExecJavaClassLoader@34fe326d.
log4j: Using URL [jar:file:/Users/karanalang/.m2/repository/io/confluent/kafka-schema-registry/7.6.0/kafka-schema-registry-7.6.0.jar!/log4j.properties] for automatic log4j configuration.
log4j: Reading configuration from URL jar:file:/Users/karanalang/.m2/repository/io/confluent/kafka-schema-registry/7.6.0/kafka-schema-registry-7.6.0.jar!/log4j.properties
log4j: Parsing for [root] with value=[INFO, stdout].
log4j: Level token is [INFO].
log4j: Category root set to INFO
log4j: Parsing appender named "stdout".
log4j: Parsing layout options for "stdout".
log4j: Setting property [conversionPattern] to [[%d] %p %m (%c:%L)%n].
log4j: End of parsing for "stdout".
log4j: Parsed "stdout" options.
log4j: Parsing for [org.apache.directory] with value=[ERROR, stdout].
log4j: Level token is [ERROR].
log4j: Category org.apache.directory set to ERROR
log4j: Parsing appender named "stdout".
log4j: Appender "stdout" was already parsed.
log4j: Handling log4j.additivity.org.apache.directory=[null]
log4j: Parsing for [org.apache.zookeeper] with value=[ERROR, stdout].
log4j: Level token is [ERROR].
log4j: Category org.apache.zookeeper set to ERROR
log4j: Parsing appender named "stdout".
log4j: Appender "stdout" was already parsed.
log4j: Handling log4j.additivity.org.apache.zookeeper=[null]
log4j: Parsing for [org.apache.kafka] with value=[ERROR, stdout].
log4j: Level token is [ERROR].
log4j: Category org.apache.kafka set to ERROR
log4j: Parsing appender named "stdout".
log4j: Appender "stdout" was already parsed.
log4j: Handling log4j.additivity.org.apache.kafka=[null]
log4j: Parsing for [kafka] with value=[ERROR, stdout].
log4j: Level token is [ERROR].
log4j: Category kafka set to ERROR
log4j: Parsing appender named "stdout".
log4j: Appender "stdout" was already parsed.
log4j: Handling log4j.additivity.kafka=[null]
log4j: Finished configuring.
SLF4J: Actual binding is of type [org.slf4j.impl.Reload4jLoggerFactory]
[2025-05-04 15:37:20,915] INFO KafkaAvroSerializerConfig values:
auto.register.schemas = false
avro.reflection.allow.null = false
avro.remove.java.properties = false
avro.use.logical.type.converters = false
basic.auth.credentials.source = URL
basic.auth.user.info = [hidden]
bearer.auth.cache.expiry.buffer.seconds = 300
bearer.auth.client.id = null
bearer.auth.client.secret = null
bearer.auth.credentials.source = USER_INFO
bearer.auth.custom.provider.class = null
bearer.auth.identity.pool.id = null
bearer.auth.issuer.endpoint.url = null
bearer.auth.logical.cluster = null
bearer.auth.scope = null
bearer.auth.scope.claim.name = scope
bearer.auth.sub.claim.name = sub
bearer.auth.token = [hidden]
context.name.strategy = class io.confluent.kafka.serializers.context.NullContextNameStrategy
http.connect.timeout.ms = 60000
http.read.timeout.ms = 60000
id.compatibility.strict = true
key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
latest.cache.size = 1000
latest.cache.ttl.sec = -1
latest.compatibility.strict = true
max.schemas.per.subject = 1000
normalize.schemas = false
proxy.host =
proxy.port = -1
rule.actions = []
rule.executors = []
rule.service.loader.enable = true
schema.format = null
schema.reflection = false
schema.registry.basic.auth.user.info = [hidden]
schema.registry.ssl.cipher.suites = null
schema.registry.ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
schema.registry.ssl.endpoint.identification.algorithm = HTTPS
schema.registry.ssl.engine.factory.class = null
schema.registry.ssl.key.password = null
schema.registry.ssl.keymanager.algorithm = SunX509
schema.registry.ssl.keystore.certificate.chain = null
schema.registry.ssl.keystore.key = null
schema.registry.ssl.keystore.location = null
schema.registry.ssl.keystore.password = null
schema.registry.ssl.keystore.type = JKS
schema.registry.ssl.protocol = TLSv1.3
schema.registry.ssl.provider = null
schema.registry.ssl.secure.random.implementation = null
schema.registry.ssl.trustmanager.algorithm = PKIX
schema.registry.ssl.truststore.certificates = null
schema.registry.ssl.truststore.location = /Users/karanalang/Documents/Technology/apicurio-schema-registry/SchemaRegistry/apicurio_confluent/apicurio-truststore.jks
schema.registry.ssl.truststore.password = [hidden]
schema.registry.ssl.truststore.type = JKS
schema.registry.url = [https://apicurio-sr.vkp.versa-vani.com/apis/ccompat/v7]
use.latest.version = true
use.latest.with.metadata = null
use.schema.id = 2
value.subject.name.strategy = class io.confluent.kafka.serializers.subject.RecordNameStrategy
(io.confluent.kafka.serializers.KafkaAvroSerializerConfig:370)
Producer created successfully. Sending messages to: syslog.ueba-nov.v1.nov.nov
Sending employee records...
Sending record for John Doe with ID: 1
[WARNING]
org.apache.kafka.common.errors.SerializationException: Error retrieving schema ID{"type":"record","name":"Employee","namespace":"com.versa.apicurio.confluent","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"salary","type":["null","float"],"default":null},{"name":"age","type":["null","int"],"default":null},{"name":"department","type":{"type":"enum","name":"DepartmentEnum","symbols":["HR","ENGINEERING","SALES"]}},{"name":"email","type":["null","string"],"default":null},{"name":"new_col","type":["null","string"],"default":null},{"name":"new_col2","type":["null","string"],"default":null},{"name":"new_col3","type":["null","string"],"default":null}]}
at io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe.toKafkaException (AbstractKafkaSchemaSerDe.java:809)
at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl (AbstractKafkaAvroSerializer.java:176)
at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize (KafkaAvroSerializer.java:68)
at org.apache.kafka.clients.producer.KafkaProducer.doSend (KafkaProducer.java:1000)
at org.apache.kafka.clients.producer.KafkaProducer.send (KafkaProducer.java:947)
at com.versa.apicurio.confluent.VersionBasedKafkaProducer.main (VersionBasedKafkaProducer.java:507)
at org.codehaus.mojo.exec.ExecJavaMojo.doMain (ExecJavaMojo.java:375)
at org.codehaus.mojo.exec.ExecJavaMojo.doExec (ExecJavaMojo.java:364)
at org.codehaus.mojo.exec.ExecJavaMojo.lambda$execute$0 (ExecJavaMojo.java:286)
at java.lang.Thread.run (Thread.java:833)
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Error; error code: 50005
at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest (RestService.java:336)
at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest (RestService.java:409)
at io.confluent.kafka.schemaregistry.client.rest.RestService.getId (RestService.java:907)
at io.confluent.kafka.schemaregistry.client.rest.RestService.getId (RestService.java:880)
at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaByIdFromRegistry (CachedSchemaRegistryClient.java:333)
at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaBySubjectAndId (CachedSchemaRegistryClient.java:464)
at io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe.getSchemaBySubjectAndId (AbstractKafkaSchemaSerDe.java:534)
at io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe.lookupSchemaBySubjectAndId (AbstractKafkaSchemaSerDe.java:540)
at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl (AbstractKafkaAvroSerializer.java:130)
at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize (KafkaAvroSerializer.java:68)
at org.apache.kafka.clients.producer.KafkaProducer.doSend (KafkaProducer.java:1000)
at org.apache.kafka.clients.producer.KafkaProducer.send (KafkaProducer.java:947)
at com.versa.apicurio.confluent.VersionBasedKafkaProducer.main (VersionBasedKafkaProducer.java:507)
at org.codehaus.mojo.exec.ExecJavaMojo.doMain (ExecJavaMojo.java:375)
at org.codehaus.mojo.exec.ExecJavaMojo.doExec (ExecJavaMojo.java:364)
at org.codehaus.mojo.exec.ExecJavaMojo.lambda$execute$0 (ExecJavaMojo.java:286)
at java.lang.Thread.run (Thread.java:833)
[INFO] ------------------------------------------------------------------------
[INFO] BUILD FAILURE
Любая идея об этом?>
Подробнее здесь: https://stackoverflow.com/questions/796 ... bility-mod
