Skip to content
This repository was archived by the owner on Jan 24, 2024. It is now read-only.

[transaction] Use the config's metadata tenant when disable multi-tenant metadata #1980

Open
Demogorgon314 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[transaction] Use the config's metadata tenant when disable multi-tenant metadata #1980

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
8 changes: 6 additions & 2 deletions kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -470,12 +470,16 @@ public ProducerStateManagerSnapshotBuffer apply(String tenant) {
if (!kafkaConfig.isKafkaTransactionCoordinatorEnabled()) {
return new MemoryProducerStateManagerSnapshotBuffer();
}
return getTransactionCoordinator(tenant)
if (kafkaConfig.isKafkaEnableMultiTenantMetadata()) {
return getTransactionCoordinator(tenant)
.getProducerStateManagerSnapshotBuffer();
}
return getTransactionCoordinator(kafkaConfig.getKafkaMetadataTenant())
.getProducerStateManagerSnapshotBuffer();
}
}

private Function<String, ProducerStateManagerSnapshotBuffer> getProducerStateManagerSnapshotBufferByTenant =
protected final Function<String, ProducerStateManagerSnapshotBuffer> getProducerStateManagerSnapshotBufferByTenant =
new ProducerStateManagerSnapshotProvider();

// this is called after initialize, and with kafkaConfig, brokerService all set.
Expand Down
45 changes: 45 additions & 0 deletions tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import static org.apache.pulsar.common.naming.TopicName.PARTITIONED_TOPIC_SUFFIX;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
Expand All @@ -38,6 +39,9 @@
import io.streamnative.pulsar.handlers.kop.coordinator.group.GroupMetadata;
import io.streamnative.pulsar.handlers.kop.coordinator.group.GroupMetadataManager;
import io.streamnative.pulsar.handlers.kop.offset.OffsetAndMetadata;
import io.streamnative.pulsar.handlers.kop.storage.MemoryProducerStateManagerSnapshotBuffer;
import io.streamnative.pulsar.handlers.kop.storage.ProducerStateManagerSnapshotBuffer;
import io.streamnative.pulsar.handlers.kop.storage.PulsarTopicProducerStateManagerSnapshotBuffer;
import io.streamnative.pulsar.handlers.kop.utils.KafkaResponseUtils;
import io.streamnative.pulsar.handlers.kop.utils.TopicNameUtils;
import java.net.InetSocketAddress;
Expand All @@ -58,10 +62,12 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewPartitions;
Expand Down Expand Up @@ -1098,6 +1104,45 @@ public void testCommitOffsetRetryWhenProducerClosed()
assertEquals(fetchMessages, numMessages);
}

@Test(timeOut = 30000)
public void testProducerStateManagerSnapshotProvider() throws IllegalAccessException {
boolean kafkaTransactionCoordinatorEnabled = conf.isKafkaTransactionCoordinatorEnabled();
boolean kafkaEnableMultiTenantMetadata = conf.isKafkaEnableMultiTenantMetadata();
try {
String tenant = "test-tenant";
KafkaProtocolHandler protocolHandler = spy(getProtocolHandler());
conf.setKafkaTransactionCoordinatorEnabled(false);
Function<String, ProducerStateManagerSnapshotBuffer> fun =
protocolHandler.getProducerStateManagerSnapshotBufferByTenant;
ProducerStateManagerSnapshotBuffer buffer = fun.apply(tenant);
assertNotNull(buffer);
assertTrue(buffer instanceof MemoryProducerStateManagerSnapshotBuffer);

// test multi-tenant
conf.setKafkaEnableMultiTenantMetadata(true);
conf.setKafkaTransactionCoordinatorEnabled(true);
buffer = fun.apply(tenant);
assertNotNull(buffer);
assertTrue(buffer instanceof PulsarTopicProducerStateManagerSnapshotBuffer);
PulsarTopicProducerStateManagerSnapshotBuffer bufferImpl =
(PulsarTopicProducerStateManagerSnapshotBuffer) buffer;
String topic = (String) FieldUtils.readField(bufferImpl, "topic", true);
assertEquals(TopicName.get(topic).getTenant(), tenant);

// test default tenant
conf.setKafkaEnableMultiTenantMetadata(false);
buffer = fun.apply(tenant);
assertNotNull(buffer);
assertTrue(buffer instanceof PulsarTopicProducerStateManagerSnapshotBuffer);
bufferImpl = (PulsarTopicProducerStateManagerSnapshotBuffer) buffer;
topic = (String) FieldUtils.readField(bufferImpl, "topic", true);
assertEquals(TopicName.get(topic).getTenant(), conf.getKafkaMetadataTenant());
} finally {
conf.setKafkaTransactionCoordinatorEnabled(kafkaTransactionCoordinatorEnabled);
conf.setKafkaEnableMultiTenantMetadata(kafkaEnableMultiTenantMetadata);
}
}

private KafkaHeaderAndRequest createTopicMetadataRequest(List<String> topics, boolean allowAutoTopicCreation) {
AbstractRequest.Builder builder = new MetadataRequest.Builder(topics, allowAutoTopicCreation);
return buildRequest(builder);
Expand Down