Skip to content

Java Client: remove usage of reflection while using Pulsar Implementation classes #11636

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merlimat merged 4 commits into from
Aug 12, 2021
Merged

Java Client: remove usage of reflection while using Pulsar Implementation classes #11636

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c42ad26
Java Client: remove usage of reflection for loading Pulsar Impl classes
eolivelli Aug 11, 2021
ba6b9f9
Fix checkstyle and build
eolivelli Aug 11, 2021
21cbb23
fix unit test
eolivelli Aug 11, 2021
9448da3
Add some javadoc and remove use of fully qualified class names
eolivelli Aug 12, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 23 additions & 11 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import static org.apache.commons.lang.StringUtils.defaultIfEmpty;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Charsets;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.Clock;
import java.util.List;
Expand Down Expand Up @@ -143,8 +144,14 @@ public void postSchema(PostSchemaPayload payload, boolean authoritative, AsyncRe
}
byte[] data;
if (SchemaType.KEY_VALUE.name().equals(payload.getType())) {
data = DefaultImplementation
.convertKeyValueDataStringToSchemaInfoSchema(payload.getSchema().getBytes(Charsets.UTF_8));
try {
data = DefaultImplementation.getDefaultImplementation()
.convertKeyValueDataStringToSchemaInfoSchema(payload.getSchema().getBytes(Charsets.UTF_8));
} catch (IOException conversionError) {
log.error("[{}] Failed to post schema for topic {}", clientAppId(), topicName, conversionError);
response.resume(Response.serverError().build());
return;
}
} else {
data = payload.getSchema().getBytes(Charsets.UTF_8);
}
Expand Down Expand Up @@ -241,16 +248,21 @@ protected String domain() {
}

private static GetSchemaResponse convertSchemaAndMetadataToGetSchemaResponse(SchemaAndMetadata schemaAndMetadata) {
String schemaData;
if (schemaAndMetadata.schema.getType() == SchemaType.KEY_VALUE) {
schemaData = DefaultImplementation.convertKeyValueSchemaInfoDataToString(
DefaultImplementation.decodeKeyValueSchemaInfo(schemaAndMetadata.schema.toSchemaInfo()));
} else {
schemaData = new String(schemaAndMetadata.schema.getData(), UTF_8);
try {
String schemaData;
if (schemaAndMetadata.schema.getType() == SchemaType.KEY_VALUE) {
schemaData = DefaultImplementation.getDefaultImplementation().convertKeyValueSchemaInfoDataToString(
DefaultImplementation.getDefaultImplementation()
.decodeKeyValueSchemaInfo(schemaAndMetadata.schema.toSchemaInfo()));
} else {
schemaData = new String(schemaAndMetadata.schema.getData(), UTF_8);
}
return GetSchemaResponse.builder().version(getLongSchemaVersion(schemaAndMetadata.version))
.type(schemaAndMetadata.schema.getType()).timestamp(schemaAndMetadata.schema.getTimestamp())
.data(schemaData).properties(schemaAndMetadata.schema.getProps()).build();
} catch (IOException conversionError) {
throw new RuntimeException(conversionError);
}
return GetSchemaResponse.builder().version(getLongSchemaVersion(schemaAndMetadata.version))
.type(schemaAndMetadata.schema.getType()).timestamp(schemaAndMetadata.schema.getTimestamp())
.data(schemaData).properties(schemaAndMetadata.schema.getProps()).build();
}

private static void handleGetSchemaResponse(AsyncResponse response, SchemaAndMetadata schema, Throwable error) {
Expand Down
4 changes: 2 additions & 2 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageIdSerializationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
package org.apache.pulsar.broker.service;

import static org.testng.Assert.assertEquals;

import java.io.IOException;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.testng.annotations.Test;
Expand All @@ -46,7 +46,7 @@ public void testProtobufSerializationNull() throws Exception {
MessageId.fromByteArray(null);
}

@Test(expectedExceptions = RuntimeException.class)
@Test(expectedExceptions = IOException.class)
void testProtobufSerializationEmpty() throws Exception {
MessageId.fromByteArray(new byte[0]);
}
Expand Down
3 changes: 1 addition & 2 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,6 @@
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.ClusterDataImpl;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
Expand Down Expand Up @@ -548,7 +547,7 @@ private void txnCumulativeAckTest(boolean batchEnable, int maxBatchSize, Subscri
}

try {
consumer.acknowledgeCumulativeAsync(DefaultImplementation
consumer.acknowledgeCumulativeAsync(DefaultImplementation.getDefaultImplementation()
.newMessageId(((MessageIdImpl) message.getMessageId()).getLedgerId(),
((MessageIdImpl) message.getMessageId()).getEntryId() - 1, -1),
abortTxn).get();
Expand Down
34 changes: 21 additions & 13 deletions pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SchemasImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.apache.pulsar.client.admin.internal;

import static java.nio.charset.StandardCharsets.UTF_8;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
Expand Down Expand Up @@ -438,8 +439,12 @@ static SchemaInfo convertGetSchemaResponseToSchemaInfo(TopicName tn,

byte[] schema;
if (response.getType() == SchemaType.KEY_VALUE) {
schema = DefaultImplementation.convertKeyValueDataStringToSchemaInfoSchema(
response.getData().getBytes(UTF_8));
try {
schema = DefaultImplementation.getDefaultImplementation().convertKeyValueDataStringToSchemaInfoSchema(
response.getData().getBytes(UTF_8));
} catch (IOException conversionError) {
throw new RuntimeException(conversionError);
}
} else {
schema = response.getData().getBytes(UTF_8);
}
Expand All @@ -466,28 +471,31 @@ static SchemaInfoWithVersion convertGetSchemaResponseToSchemaInfoWithVersion(Top


// the util function exists for backward compatibility concern
static String convertSchemaDataToStringLegacy(SchemaInfo schemaInfo) {
static String convertSchemaDataToStringLegacy(SchemaInfo schemaInfo) throws IOException {
byte[] schemaData = schemaInfo.getSchema();
if (null == schemaInfo.getSchema()) {
return "";
}

if (schemaInfo.getType() == SchemaType.KEY_VALUE) {
return DefaultImplementation.convertKeyValueSchemaInfoDataToString(
DefaultImplementation.decodeKeyValueSchemaInfo(schemaInfo));
return DefaultImplementation.getDefaultImplementation().convertKeyValueSchemaInfoDataToString(
DefaultImplementation.getDefaultImplementation().decodeKeyValueSchemaInfo(schemaInfo));
}

return new String(schemaData, UTF_8);
}

static PostSchemaPayload convertSchemaInfoToPostSchemaPayload(SchemaInfo schemaInfo) {

PostSchemaPayload payload = new PostSchemaPayload();
payload.setType(schemaInfo.getType().name());
payload.setProperties(schemaInfo.getProperties());
// for backward compatibility concern, we convert `bytes` to `string`
// we can consider fixing it in a new version of rest endpoint
payload.setSchema(convertSchemaDataToStringLegacy(schemaInfo));
return payload;
try {
PostSchemaPayload payload = new PostSchemaPayload();
payload.setType(schemaInfo.getType().name());
payload.setProperties(schemaInfo.getProperties());
// for backward compatibility concern, we convert `bytes` to `string`
// we can consider fixing it in a new version of rest endpoint
payload.setSchema(convertSchemaDataToStringLegacy(schemaInfo));
return payload;
} catch (IOException conversionError) {
throw new RuntimeException(conversionError);
}
}
}
10 changes: 5 additions & 5 deletions pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AuthenticationFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ public final class AuthenticationFactory {
* @return the Authentication object initialized with the token credentials
*/
public static Authentication token(String token) {
return DefaultImplementation.newAuthenticationToken(token);
return DefaultImplementation.getDefaultImplementation().newAuthenticationToken(token);
}

/**
Expand All @@ -52,7 +52,7 @@ public static Authentication token(String token) {
* @return the Authentication object initialized with the token credentials
*/
public static Authentication token(Supplier<String> tokenSupplier) {
return DefaultImplementation.newAuthenticationToken(tokenSupplier);
return DefaultImplementation.getDefaultImplementation().newAuthenticationToken(tokenSupplier);
}

// CHECKSTYLE.OFF: MethodName
Expand All @@ -67,7 +67,7 @@ public static Authentication token(Supplier<String> tokenSupplier) {
* @return the Authentication object initialized with the TLS credentials
*/
public static Authentication TLS(String certFilePath, String keyFilePath) {
return DefaultImplementation.newAuthenticationTLS(certFilePath, keyFilePath);
return DefaultImplementation.getDefaultImplementation().newAuthenticationTLS(certFilePath, keyFilePath);
}

// CHECKSTYLE.ON: MethodName
Expand All @@ -86,7 +86,7 @@ public static Authentication TLS(String certFilePath, String keyFilePath) {
public static Authentication create(String authPluginClassName, String authParamsString)
throws UnsupportedAuthenticationException {
try {
return DefaultImplementation.createAuthentication(authPluginClassName, authParamsString);
return DefaultImplementation.getDefaultImplementation().createAuthentication(authPluginClassName, authParamsString);
} catch (Throwable t) {
throw new UnsupportedAuthenticationException(t);
}
Expand All @@ -103,7 +103,7 @@ public static Authentication create(String authPluginClassName, String authParam
public static Authentication create(String authPluginClassName, Map<String, String> authParams)
throws UnsupportedAuthenticationException {
try {
return DefaultImplementation.createAuthentication(authPluginClassName, authParams);
return DefaultImplementation.getDefaultImplementation().createAuthentication(authPluginClassName, authParams);
} catch (Throwable t) {
throw new UnsupportedAuthenticationException(t);
}
Expand Down
5 changes: 3 additions & 2 deletions pulsar-client-api/src/main/java/org/apache/pulsar/client/api/BatcherBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.apache.pulsar.client.api;

import java.io.Serializable;

import org.apache.pulsar.client.internal.DefaultImplementation;
import org.apache.pulsar.common.classification.InterfaceAudience;
import org.apache.pulsar.common.classification.InterfaceStability;
Expand All @@ -39,7 +40,7 @@ public interface BatcherBuilder extends Serializable {
* <p>batched into single batch message:
* [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
*/
BatcherBuilder DEFAULT = DefaultImplementation.newDefaultBatcherBuilder();
BatcherBuilder DEFAULT = DefaultImplementation.getDefaultImplementation().newDefaultBatcherBuilder();

/**
* Key based batch message container.
Expand All @@ -50,7 +51,7 @@ public interface BatcherBuilder extends Serializable {
* <p>batched into multiple batch messages:
* [(k1, v1), (k1, v2), (k1, v3)], [(k2, v1), (k2, v2), (k2, v3)], [(k3, v1), (k3, v2), (k3, v3)]
*/
BatcherBuilder KEY_BASED = DefaultImplementation.newKeyBasedBatcherBuilder();
BatcherBuilder KEY_BASED = DefaultImplementation.getDefaultImplementation().newKeyBasedBatcherBuilder();

/**
* Build a new batch message container.
Expand Down
9 changes: 5 additions & 4 deletions pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MessageId.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

import java.io.IOException;
import java.io.Serializable;

import org.apache.pulsar.client.internal.DefaultImplementation;
import org.apache.pulsar.common.classification.InterfaceAudience;
import org.apache.pulsar.common.classification.InterfaceStability;
Expand Down Expand Up @@ -54,7 +55,7 @@ public interface MessageId extends Comparable<MessageId>, Serializable {
* @throws IOException if the de-serialization fails
*/
static MessageId fromByteArray(byte[] data) throws IOException {
return DefaultImplementation.newMessageIdFromByteArray(data);
return DefaultImplementation.getDefaultImplementation().newMessageIdFromByteArray(data);
}

/**
Expand All @@ -70,20 +71,20 @@ static MessageId fromByteArray(byte[] data) throws IOException {
* @throws IOException if the de-serialization fails
*/
static MessageId fromByteArrayWithTopic(byte[] data, String topicName) throws IOException {
return DefaultImplementation.newMessageIdFromByteArrayWithTopic(data, topicName);
return DefaultImplementation.getDefaultImplementation().newMessageIdFromByteArrayWithTopic(data, topicName);
}

// CHECKSTYLE.OFF: ConstantName

/**
* MessageId that represents the oldest message available in the topic.
*/
MessageId earliest = DefaultImplementation.newMessageId(-1, -1, -1);
MessageId earliest = DefaultImplementation.getDefaultImplementation().newMessageId(-1, -1, -1);

/**
* MessageId that represents the next message published in the topic.
*/
MessageId latest = DefaultImplementation.newMessageId(Long.MAX_VALUE, Long.MAX_VALUE, -1);
MessageId latest = DefaultImplementation.getDefaultImplementation().newMessageId(Long.MAX_VALUE, Long.MAX_VALUE, -1);

// CHECKSTYLE.ON: ConstantName
}
2 changes: 1 addition & 1 deletion pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ public interface PulsarClient extends Closeable {
* @since 2.0.0
*/
static ClientBuilder builder() {
return DefaultImplementation.newClientBuilder();
return DefaultImplementation.getDefaultImplementation().newClientBuilder();
}

/**
Expand Down
Loading