Skip to content
Closed
Original file line number Diff line number Diff line change
Expand Up @@ -28,12 +28,13 @@
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;

import com.google.common.collect.Sets;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.bookkeeper.mledger.LedgerOffloader;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
Expand Down Expand Up @@ -104,12 +105,12 @@ private void testOffload(String topicName, String mlName) throws Exception {
ManagedLedgerInfo info = pulsar.getManagedLedgerFactory().getManagedLedgerInfo(mlName);
assertEquals(info.ledgers.size(), 2);

assertEquals(admin.topics().offloadStatus(topicName).status,
assertEquals(admin.topics().offloadStatus(topicName).getStatus(),
LongRunningProcessStatus.Status.NOT_RUN);

admin.topics().triggerOffload(topicName, currentId);

assertEquals(admin.topics().offloadStatus(topicName).status,
assertEquals(admin.topics().offloadStatus(topicName).getStatus(),
LongRunningProcessStatus.Status.RUNNING);

try {
Expand All @@ -122,9 +123,9 @@ private void testOffload(String topicName, String mlName) throws Exception {
// fail first time
promise.completeExceptionally(new Exception("Some random failure"));

assertEquals(admin.topics().offloadStatus(topicName).status,
assertEquals(admin.topics().offloadStatus(topicName).getStatus(),
LongRunningProcessStatus.Status.ERROR);
Assert.assertTrue(admin.topics().offloadStatus(topicName).lastError.contains("Some random failure"));
Assert.assertTrue(admin.topics().offloadStatus(topicName).getLastError().contains("Some random failure"));

// Try again
doReturn(CompletableFuture.completedFuture(null))
Expand All @@ -133,12 +134,14 @@ private void testOffload(String topicName, String mlName) throws Exception {
admin.topics().triggerOffload(topicName, currentId);

Awaitility.await().untilAsserted(() ->
assertEquals(admin.topics().offloadStatus(topicName).status,
assertEquals(admin.topics().offloadStatus(topicName).getStatus(),
LongRunningProcessStatus.Status.SUCCESS));
MessageIdImpl firstUnoffloaded = admin.topics().offloadStatus(topicName).firstUnoffloadedMessage;
MessageId firstUnoffloaded = admin.topics().offloadStatus(topicName).getFirstUnoffloadedMessage();
assertTrue(firstUnoffloaded instanceof MessageIdImpl);
MessageIdImpl firstUnoffloadedMessage = (MessageIdImpl) firstUnoffloaded;
// First unoffloaded is the first entry of current ledger
assertEquals(firstUnoffloaded.getLedgerId(), info.ledgers.get(1).ledgerId);
assertEquals(firstUnoffloaded.getEntryId(), 0);
assertEquals(firstUnoffloadedMessage.getLedgerId(), info.ledgers.get(1).ledgerId);
assertEquals(firstUnoffloadedMessage.getEntryId(), 0);

verify(offloader, times(2)).offload(any(), any(), any());
}
Expand Down
9 changes: 7 additions & 2 deletions pulsar-client-admin-api/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
<name>Pulsar Client Admin :: API</name>

<dependencies>

<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-common</artifactId>
Expand All @@ -41,10 +42,14 @@

<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-client-original</artifactId>
<version>${project.parent.version}</version>
<artifactId>pulsar-package-core</artifactId>
<version>${project.version}</version>
</dependency>

<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if gson is not already a dependency imported by pulsar-client-original in 2.7.x I would like to not add it.
Users of pulsar-client-original already have Jackson and usually who uses pulsar-client-admin-api uses pulsar-client, so I would lean toward using only Jackson.

what's your plan for removing gson in a follow up patch ? this next patch will become a blocker for the 2.8 release

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Adding gson as the dependency is a temporary action, with removal of pulsar-client-original and pulsar-package-core, pulsar-client-admin-api will lose dependency of gson, but we still have a lot of classes import gson, such as

.

I do understand that the 2.8 release is in a hurry, but there are indeed many problems that need to be fixed one by one in order to achieve the ultimate goal. So the removal of gson will be in a separate PR soon.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My problem was about "adding" gson, if it was not already there, so to me this is not a problem.

thanks for your clarification

</dependency>
</dependencies>

<build>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,38 +18,30 @@
*/
package org.apache.pulsar.client.admin;

import org.apache.pulsar.client.admin.utils.DefaultImplementation;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.impl.MessageIdImpl;

/**
* Status of offload process.
* interface class of Status of offload process.
*/
public class OffloadProcessStatus extends LongRunningProcessStatus {
public interface OffloadProcessStatus {

public MessageIdImpl firstUnoffloadedMessage;
MessageId getFirstUnoffloadedMessage();
String getLastError();
LongRunningProcessStatus.Status getStatus();

public OffloadProcessStatus() {
super(Status.NOT_RUN, "");
firstUnoffloadedMessage = (MessageIdImpl) MessageId.earliest;
static OffloadProcessStatus forStatus(LongRunningProcessStatus.Status status) {
return DefaultImplementation.newOffloadProcessStatus(status, "", MessageId.earliest);
}

private OffloadProcessStatus(Status status, String lastError,
MessageIdImpl firstUnoffloadedMessage) {
this.status = status;
this.lastError = lastError;
this.firstUnoffloadedMessage = firstUnoffloadedMessage;
static OffloadProcessStatus forError(String lastError) {
return DefaultImplementation.newOffloadProcessStatus(LongRunningProcessStatus.Status.ERROR, lastError,
MessageId.earliest);
}

public static OffloadProcessStatus forStatus(Status status) {
return new OffloadProcessStatus(status, "", (MessageIdImpl) MessageId.earliest);
static OffloadProcessStatus forSuccess(MessageId messageId) {
return DefaultImplementation.newOffloadProcessStatus(LongRunningProcessStatus.Status.SUCCESS, "",
messageId);
}

public static OffloadProcessStatus forError(String lastError) {
return new OffloadProcessStatus(Status.ERROR, lastError,
(MessageIdImpl) MessageId.earliest);
}

public static OffloadProcessStatus forSuccess(MessageIdImpl messageId) {
return new OffloadProcessStatus(Status.SUCCESS, "", messageId);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@

import java.io.Closeable;
import org.apache.pulsar.client.admin.utils.DefaultImplementation;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.common.classification.InterfaceAudience;
import org.apache.pulsar.common.classification.InterfaceStability;

Expand Down Expand Up @@ -147,11 +146,6 @@ static PulsarAdminBuilder builder() {
*/
String getServiceUrl();

/**
* @return the client Configuration Data that is being used
*/
ClientConfigurationData getClientConfigData();

/**
* @return the schemas
*/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,12 @@
*/
package org.apache.pulsar.client.admin.utils;

import java.lang.reflect.Constructor;
import lombok.experimental.UtilityClass;
import org.apache.pulsar.client.admin.LongRunningProcessStatus;
import org.apache.pulsar.client.admin.OffloadProcessStatus;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.MessageId;

/**
* Helper class for class instantiations and it also contains methods to work with schemas.
Expand All @@ -30,7 +34,18 @@ public class DefaultImplementation {
private static final Class<PulsarAdminBuilder> ADMIN_CLIENT_BUILDER_IMPL = ReflectionUtils.newClassInstance(
"org.apache.pulsar.client.admin.internal.PulsarAdminBuilderImpl");

private static final Constructor<OffloadProcessStatus> OFFLOAD_PROCESS_STATUS_IMPL_status_string_messageid =
ReflectionUtils.getConstructor(
"org.apache.pulsar.client.admin.internal.OffloadProcessStatusImpl",
LongRunningProcessStatus.Status.class, String.class, MessageId.class);

public static PulsarAdminBuilder newAdminClientBuilder() {
return ReflectionUtils.catchExceptions(() -> ADMIN_CLIENT_BUILDER_IMPL.newInstance());
}

public static OffloadProcessStatus newOffloadProcessStatus(LongRunningProcessStatus.Status status, String lastError
, MessageId messageId) {
return ReflectionUtils.catchExceptions(() -> OFFLOAD_PROCESS_STATUS_IMPL_status_string_messageid.newInstance(
status, lastError, messageId));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,11 @@
package org.apache.pulsar.client.admin.internal;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.module.SimpleAbstractTypeResolver;
import com.fasterxml.jackson.databind.module.SimpleModule;
import javax.ws.rs.ext.ContextResolver;
import javax.ws.rs.ext.Provider;
import org.apache.pulsar.client.admin.OffloadProcessStatus;
import org.apache.pulsar.common.util.ObjectMapperFactory;

/**
Expand All @@ -33,6 +36,20 @@ public class JacksonConfigurator implements ContextResolver<ObjectMapper> {

public JacksonConfigurator() {
mapper = ObjectMapperFactory.create();
setInterfaceDefaultMapperModule();
}

private void setInterfaceDefaultMapperModule() {
SimpleModule module = new SimpleModule("InterfaceDefaultMapperModule");

// we not specific @JsonDeserialize annotation in OffloadProcessStatus
// because we do not want to have jackson dependency in pulsar-client-admin-api
// In this case we use SimpleAbstractTypeResolver to map interfaces to impls
SimpleAbstractTypeResolver resolver = new SimpleAbstractTypeResolver();
resolver.addMapping(OffloadProcessStatus.class, OffloadProcessStatusImpl.class);
Copy link
Contributor Author

@freeznet freeznet May 12, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@merlimat I have added a SimpleAbstractTypeResolver here, and will manually map interfaces with impl classes.


module.setAbstractTypes(resolver);
mapper.registerModule(module);
}

@Override
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
/**
* 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.pulsar.client.admin.internal;

import org.apache.pulsar.client.admin.LongRunningProcessStatus;
import org.apache.pulsar.client.admin.OffloadProcessStatus;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.impl.MessageIdImpl;

/**
* Status of offload process.
*/
public class OffloadProcessStatusImpl extends LongRunningProcessStatus implements OffloadProcessStatus {
public MessageIdImpl firstUnoffloadedMessage;

public OffloadProcessStatusImpl() {
status = Status.NOT_RUN;
lastError = "";
firstUnoffloadedMessage = (MessageIdImpl) MessageId.earliest;
}

public OffloadProcessStatusImpl(Status status, String lastError,
MessageId firstUnoffloadedMessage) {
this.status = status;
this.lastError = lastError;
this.firstUnoffloadedMessage = (MessageIdImpl) firstUnoffloadedMessage;
}

@Override
public MessageId getFirstUnoffloadedMessage() {
return firstUnoffloadedMessage;
}

@Override
public String getLastError() {
return lastError;
}

@Override
public Status getStatus() {
return status;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,6 @@ public void setup() throws Exception {
this.functions = mock(Functions.class);
when(admin.functions()).thenReturn(functions);
when(admin.getServiceUrl()).thenReturn("http://localhost:1234");
when(admin.getClientConfigData()).thenReturn(new ClientConfigurationData());
this.cmd = new CmdFunctions(() -> admin);
this.cmdSinks = new CmdSinks(() -> admin);
this.cmdSources = new CmdSources(() -> admin);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@
import org.apache.pulsar.client.admin.Schemas;
import org.apache.pulsar.client.admin.Tenants;
import org.apache.pulsar.client.admin.Topics;
import org.apache.pulsar.client.admin.internal.OffloadProcessStatusImpl;
import org.apache.pulsar.client.admin.internal.PulsarAdminBuilderImpl;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.SubscriptionType;
Expand Down Expand Up @@ -1113,7 +1114,7 @@ public boolean matches(Long timestamp) {
cmdTopics.run(split("offload persistent://myprop/clust/ns1/ds1 -s 1k"));
verify(mockTopics).triggerOffload("persistent://myprop/clust/ns1/ds1", new MessageIdImpl(2, 0, -1));

when(mockTopics.offloadStatus("persistent://myprop/clust/ns1/ds1")).thenReturn(new OffloadProcessStatus());
when(mockTopics.offloadStatus("persistent://myprop/clust/ns1/ds1")).thenReturn(new OffloadProcessStatusImpl());
cmdTopics.run(split("offload-status persistent://myprop/clust/ns1/ds1"));
verify(mockTopics).offloadStatus("persistent://myprop/clust/ns1/ds1");

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
import java.util.stream.Collectors;

import org.apache.pulsar.client.admin.LongRunningProcessStatus;
import org.apache.pulsar.client.admin.OffloadProcessStatus;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.admin.Topics;
Expand Down Expand Up @@ -1035,13 +1036,13 @@ void run() throws PulsarAdminException {
String persistentTopic = validatePersistentTopic(params);

try {
LongRunningProcessStatus status = getTopics().offloadStatus(persistentTopic);
while (wait && status.status == LongRunningProcessStatus.Status.RUNNING) {
OffloadProcessStatus status = getTopics().offloadStatus(persistentTopic);
while (wait && status.getStatus() == LongRunningProcessStatus.Status.RUNNING) {
Thread.sleep(1000);
status = getTopics().offloadStatus(persistentTopic);
}

switch (status.status) {
switch (status.getStatus()) {
case NOT_RUN:
System.out.println("Offload has not been run for " + persistentTopic
+ " since broker startup");
Expand All @@ -1054,7 +1055,7 @@ void run() throws PulsarAdminException {
break;
case ERROR:
System.out.println("Error in offload");
throw new PulsarAdminException("Error offloading: " + status.lastError);
throw new PulsarAdminException("Error offloading: " + status.getLastError());
}
} catch (InterruptedException e) {
throw new PulsarAdminException(e);
Expand Down
2 changes: 1 addition & 1 deletion pulsar-functions/runtime-all/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@

<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-client-api</artifactId>
<artifactId>pulsar-client-original</artifactId>
<version>${project.version}</version>
</dependency>

Expand Down
6 changes: 6 additions & 0 deletions pulsar-io/kafka/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,12 @@
<scope>provided</scope>
</dependency>

<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-client-original</artifactId>
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we need pulsar client explicitly for Kafka connector?

Copy link
Contributor Author

@freeznet freeznet May 11, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@merlimat I believe this is related to the conversion between pulsar's schema and kafka's schema. So kafka connector need pulsar-client-original which all schema impls are in the package.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, @freeznet is right. we need Schema Impl classes

<version>${project.version}</version>
</dependency>

<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
Expand Down
5 changes: 5 additions & 0 deletions tests/docker-images/java-test-functions/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,11 @@
<artifactId>pulsar-io-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-client-original</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
<packaging>jar</packaging>

Expand Down