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

Commit 6cbbc5b

Browse files
ZiyaoWeiZiyaoWei
committed
Handle creating topic with migration metadata and Kafka produce/fetch requests before migration
1 parent 0c72c46 commit 6cbbc5b

14 files changed

+293
-98
lines changed

‎kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaChannelInitializer.java

+9-5
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import io.netty.handler.codec.LengthFieldPrepender;
2323
import io.netty.handler.ssl.SslHandler;
2424
import io.netty.handler.timeout.IdleStateHandler;
25+
import io.streamnative.pulsar.handlers.kop.migration.metadata.MigrationMetadataManager;
2526
import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperation;
2627
import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperationPurgatory;
2728
import io.streamnative.pulsar.handlers.kop.utils.ssl.SSLUtils;
@@ -50,8 +51,9 @@ public class KafkaChannelInitializer extends ChannelInitializer<SocketChannel> {
5051
private final KafkaTopicManagerSharedState kafkaTopicManagerSharedState;
5152

5253
private final AdminManager adminManager;
53-
private DelayedOperationPurgatory<DelayedOperation> producePurgatory;
54-
private DelayedOperationPurgatory<DelayedOperation> fetchPurgatory;
54+
private final MigrationMetadataManager migrationMetadataManager;
55+
private final DelayedOperationPurgatory<DelayedOperation> producePurgatory;
56+
private final DelayedOperationPurgatory<DelayedOperation> fetchPurgatory;
5557
@Getter
5658
private final boolean enableTls;
5759
@Getter
@@ -77,7 +79,8 @@ public KafkaChannelInitializer(PulsarService pulsarService,
7779
boolean skipMessagesWithoutIndex,
7880
RequestStats requestStats,
7981
OrderedScheduler sendResponseScheduler,
80-
KafkaTopicManagerSharedState kafkaTopicManagerSharedState) {
82+
KafkaTopicManagerSharedState kafkaTopicManagerSharedState,
83+
MigrationMetadataManager migrationMetadataManager) {
8184
super();
8285
this.pulsarService = pulsarService;
8386
this.kafkaConfig = kafkaConfig;
@@ -98,6 +101,7 @@ public KafkaChannelInitializer(PulsarService pulsarService,
98101
this.sendResponseScheduler = sendResponseScheduler;
99102
this.kafkaTopicManagerSharedState = kafkaTopicManagerSharedState;
100103
this.lengthFieldPrepender = new LengthFieldPrepender(4);
104+
this.migrationMetadataManager = migrationMetadataManager;
101105
}
102106

103107
@Override
@@ -123,7 +127,7 @@ public KafkaRequestHandler newCnx() throws Exception {
123127
tenantContextManager, kopBrokerLookupManager, adminManager,
124128
producePurgatory, fetchPurgatory,
125129
enableTls, advertisedEndPoint, skipMessagesWithoutIndex, requestStats, sendResponseScheduler,
126-
kafkaTopicManagerSharedState);
130+
kafkaTopicManagerSharedState, migrationMetadataManager);
127131
}
128132

129133
@VisibleForTesting
@@ -133,6 +137,6 @@ public KafkaRequestHandler newCnxWithoutStats(final TenantContextManager tenantC
133137
producePurgatory, fetchPurgatory,
134138
enableTls, advertisedEndPoint, skipMessagesWithoutIndex, RequestStats.NULL_INSTANCE,
135139
sendResponseScheduler,
136-
kafkaTopicManagerSharedState);
140+
kafkaTopicManagerSharedState, migrationMetadataManager);
137141
}
138142
}

‎kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java

+13-6
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@
2929
import io.streamnative.pulsar.handlers.kop.format.EntryFormatterFactory;
3030
import io.streamnative.pulsar.handlers.kop.http.HttpChannelInitializer;
3131
import io.streamnative.pulsar.handlers.kop.migration.MigrationManager;
32+
import io.streamnative.pulsar.handlers.kop.migration.metadata.ManagedLedgerPropertiesMigrationMetadataManager;
33+
import io.streamnative.pulsar.handlers.kop.migration.metadata.MigrationMetadataManager;
3234
import io.streamnative.pulsar.handlers.kop.schemaregistry.SchemaRegistryChannelInitializer;
3335
import io.streamnative.pulsar.handlers.kop.stats.PrometheusMetricsProvider;
3436
import io.streamnative.pulsar.handlers.kop.stats.StatsLogger;
@@ -105,6 +107,7 @@ public class KafkaProtocolHandler implements ProtocolHandler, TenantContextManag
105107
private NamespaceBundleOwnershipListenerImpl bundleListener;
106108
private SchemaRegistryManager schemaRegistryManager;
107109
private MigrationManager migrationManager;
110+
private MigrationMetadataManager migrationMetadataManager;
108111

109112
private final Map<String, GroupCoordinator> groupCoordinatorsByTenant = new ConcurrentHashMap<>();
110113
private final Map<String, TransactionCoordinator> transactionCoordinatorByTenant = new ConcurrentHashMap<>();
@@ -208,10 +211,11 @@ public void start(BrokerService service) {
208211
KopVersion.getBuildTime());
209212

210213
brokerService = service;
214+
PulsarService pulsarService = brokerService.getPulsar();
211215
kafkaTopicManagerSharedState = new KafkaTopicManagerSharedState(brokerService);
212216
PulsarAdmin pulsarAdmin;
213217
try {
214-
pulsarAdmin = brokerService.getPulsar().getAdminClient();
218+
pulsarAdmin = pulsarService.getAdminClient();
215219
adminManager = new AdminManager(pulsarAdmin, kafkaConfig);
216220
} catch (PulsarServerException e) {
217221
log.error("Failed to get pulsarAdmin", e);
@@ -223,7 +227,7 @@ public void start(BrokerService service) {
223227
txnTopicClient = new SystemTopicClient(brokerService.pulsar(), kafkaConfig);
224228

225229
try {
226-
kopBrokerLookupManager = new KopBrokerLookupManager(kafkaConfig, brokerService.getPulsar());
230+
kopBrokerLookupManager = new KopBrokerLookupManager(kafkaConfig, pulsarService);
227231
} catch (Exception ex) {
228232
log.error("Failed to get kopBrokerLookupManager", ex);
229233
throw new IllegalStateException(ex);
@@ -265,7 +269,7 @@ private void invalidateBundleCache(TopicName topicName) {
265269

266270
// init KopEventManager
267271
kopEventManager = new KopEventManager(adminManager,
268-
brokerService.getPulsar().getLocalMetadataStore(),
272+
pulsarService.getLocalMetadataStore(),
269273
requestStats.getStatsLogger(),
270274
kafkaConfig,
271275
groupCoordinatorsByTenant);
@@ -281,9 +285,11 @@ private void invalidateBundleCache(TopicName topicName) {
281285
conf.addProperty("cluster", kafkaConfig.getClusterName());
282286
statsProvider.start(conf);
283287
brokerService.pulsar().addPrometheusRawMetricsProvider(statsProvider);
284-
schemaRegistryManager = new SchemaRegistryManager(kafkaConfig, brokerService.getPulsar(),
288+
schemaRegistryManager = new SchemaRegistryManager(kafkaConfig, pulsarService,
285289
brokerService.getAuthenticationService());
286-
migrationManager = new MigrationManager(kafkaConfig, brokerService.getPulsar());
290+
migrationMetadataManager = new ManagedLedgerPropertiesMigrationMetadataManager(
291+
new KafkaTopicLookupService(pulsarService.getBrokerService()), adminManager);
292+
migrationManager = new MigrationManager(kafkaConfig, brokerService.getPulsar(), migrationMetadataManager);
287293
}
288294

289295
private TransactionCoordinator createAndBootTransactionCoordinator(String tenant) {
@@ -408,7 +414,8 @@ private KafkaChannelInitializer newKafkaChannelInitializer(final EndPoint endPoi
408414
kafkaConfig.isSkipMessagesWithoutIndex(),
409415
requestStats,
410416
sendResponseScheduler,
411-
kafkaTopicManagerSharedState);
417+
kafkaTopicManagerSharedState,
418+
migrationMetadataManager);
412419
}
413420

414421
// this is called after initialize, and with kafkaConfig, brokerService all set.

‎kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java

+61-29
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,9 @@
3131
import io.streamnative.pulsar.handlers.kop.exceptions.KoPTopicException;
3232
import io.streamnative.pulsar.handlers.kop.format.EntryFormatter;
3333
import io.streamnative.pulsar.handlers.kop.format.EntryFormatterFactory;
34+
import io.streamnative.pulsar.handlers.kop.migration.metadata.MigrationMetadata;
35+
import io.streamnative.pulsar.handlers.kop.migration.metadata.MigrationMetadataManager;
36+
import io.streamnative.pulsar.handlers.kop.migration.metadata.MigrationStatus;
3437
import io.streamnative.pulsar.handlers.kop.offset.OffsetAndMetadata;
3538
import io.streamnative.pulsar.handlers.kop.offset.OffsetMetadata;
3639
import io.streamnative.pulsar.handlers.kop.security.SaslAuthenticator;
@@ -91,6 +94,8 @@
9194
import org.apache.commons.collections4.ListUtils;
9295
import org.apache.commons.lang3.NotImplementedException;
9396
import org.apache.commons.lang3.tuple.Pair;
97+
import org.apache.kafka.clients.producer.Producer;
98+
import org.apache.kafka.clients.producer.ProducerRecord;
9499
import org.apache.kafka.common.Node;
95100
import org.apache.kafka.common.TopicPartition;
96101
import org.apache.kafka.common.acl.AclOperation;
@@ -105,6 +110,7 @@
105110
import org.apache.kafka.common.record.InvalidRecordException;
106111
import org.apache.kafka.common.record.MemoryRecords;
107112
import org.apache.kafka.common.record.MutableRecordBatch;
113+
import org.apache.kafka.common.record.Record;
108114
import org.apache.kafka.common.record.RecordBatch;
109115
import org.apache.kafka.common.requests.AbstractRequest;
110116
import org.apache.kafka.common.requests.AbstractResponse;
@@ -193,6 +199,7 @@ public class KafkaRequestHandler extends KafkaCommandDecoder {
193199
private final ScheduledExecutorService executor;
194200
private final PulsarAdmin admin;
195201
private final MetadataStoreExtended metadataStore;
202+
private final MigrationMetadataManager migrationMetadataManager;
196203
private final SaslAuthenticator authenticator;
197204
private final Authorizer authorizer;
198205
private final AdminManager adminManager;
@@ -235,7 +242,7 @@ private String getCurrentTenant(String defaultTenant) {
235242
&& authenticator.session() != null
236243
&& authenticator.session().getPrincipal() != null
237244
&& authenticator.session().getPrincipal().getTenantSpec() != null) {
238-
String tenantSpec = authenticator.session().getPrincipal().getTenantSpec();
245+
String tenantSpec = authenticator.session().getPrincipal().getTenantSpec();
239246
return extractTenantFromTenantSpec(tenantSpec);
240247
}
241248
// fallback to using system (default) tenant
@@ -291,7 +298,8 @@ public KafkaRequestHandler(PulsarService pulsarService,
291298
boolean skipMessagesWithoutIndex,
292299
RequestStats requestStats,
293300
OrderedScheduler sendResponseScheduler,
294-
KafkaTopicManagerSharedState kafkaTopicManagerSharedState) throws Exception {
301+
KafkaTopicManagerSharedState kafkaTopicManagerSharedState,
302+
MigrationMetadataManager migrationMetadataManager) throws Exception {
295303
super(requestStats, kafkaConfig, sendResponseScheduler);
296304
this.pulsarService = pulsarService;
297305
this.tenantContextManager = tenantContextManager;
@@ -300,6 +308,7 @@ public KafkaRequestHandler(PulsarService pulsarService,
300308
this.executor = pulsarService.getExecutor();
301309
this.admin = pulsarService.getAdminClient();
302310
this.metadataStore = pulsarService.getLocalMetadataStore();
311+
this.migrationMetadataManager = migrationMetadataManager;
303312
final boolean authenticationEnabled = pulsarService.getBrokerService().isAuthenticationEnabled()
304313
&& !kafkaConfig.getSaslAllowedMechanisms().isEmpty();
305314
this.authenticator = authenticationEnabled
@@ -801,6 +810,16 @@ private void completeSendOperationForThrottling(long msgSize) {
801810
}
802811
}
803812

813+
private void produceToKafka(String topic, MemoryRecords records, MigrationMetadata migrationMetadata) {
814+
Producer<String, ByteBuffer> producer =
815+
migrationMetadataManager.getKafkaProducerForTopic(topic, "public/default",
816+
migrationMetadata.getKafkaClusterAddress());
817+
for (Record record : records.records()) {
818+
producer.send(new ProducerRecord<>(topic, record.value()));
819+
}
820+
producer.flush();
821+
}
822+
804823
@Override
805824
protected void handleProduceRequest(KafkaHeaderAndRequest produceHar,
806825
CompletableFuture<AbstractResponse> resultFuture) {
@@ -814,6 +833,7 @@ protected void handleProduceRequest(KafkaHeaderAndRequest produceHar,
814833
}
815834
final Map<TopicPartition, PartitionResponse> unauthorizedTopicResponsesMap = new ConcurrentHashMap<>();
816835
final Map<TopicPartition, PartitionResponse> invalidRequestResponses = new HashMap<>();
836+
final Map<TopicPartition, PartitionResponse> migrationInProgressResponses = new HashMap<>();
817837
final Map<TopicPartition, MemoryRecords> authorizedRequestInfo = new ConcurrentHashMap<>();
818838
int timeoutMs = produceRequest.timeout();
819839
String namespacePrefix = currentNamespacePrefix();
@@ -848,43 +868,55 @@ protected void handleProduceRequest(KafkaHeaderAndRequest produceHar,
848868
mergedResponse.putAll(response);
849869
mergedResponse.putAll(unauthorizedTopicResponsesMap);
850870
mergedResponse.putAll(invalidRequestResponses);
871+
mergedResponse.putAll(migrationInProgressResponses);
851872
resultFuture.complete(new ProduceResponse(mergedResponse));
852873
});
853874
}
854875
};
855876

856877
produceRequest.partitionRecordsOrFail().forEach((topicPartition, records) -> {
857-
try {
858-
validateRecords(produceHar.getRequest().version(), records);
859-
} catch (ApiException ex) {
860-
invalidRequestResponses.put(topicPartition,
861-
new ProduceResponse.PartitionResponse(Errors.forException(ex)));
862-
completeOne.run();
863-
return;
864-
}
865-
final String fullPartitionName = KopTopic.toString(topicPartition, namespacePrefix);
866-
authorize(AclOperation.WRITE, Resource.of(ResourceType.TOPIC, fullPartitionName))
867-
.whenCompleteAsync((isAuthorized, ex) -> {
868-
if (ex != null) {
869-
log.error("Write topic authorize failed, topic - {}. {}",
870-
fullPartitionName, ex.getMessage());
871-
unauthorizedTopicResponsesMap.put(topicPartition,
872-
new ProduceResponse.PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED));
873-
completeOne.run();
874-
return;
878+
migrationMetadataManager.getMigrationMetadata(topicPartition.topic(), "public/default", ctx.channel())
879+
.thenAccept(migrationMetadata -> {
880+
MigrationStatus migrationStatus = migrationMetadata.getMigrationStatus();
881+
if (migrationStatus == MigrationStatus.NOT_STARTED) {
882+
produceToKafka(topicPartition.topic(), records, migrationMetadata);
883+
} else if (migrationStatus == MigrationStatus.STARTED) {
884+
migrationInProgressResponses.put(topicPartition,
885+
new ProduceResponse.PartitionResponse(
886+
Errors.forCode(Errors.REBALANCE_IN_PROGRESS.code())));
875887
}
876-
if (!isAuthorized) {
877-
unauthorizedTopicResponsesMap.put(topicPartition,
878-
new ProduceResponse.PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED));
888+
try {
889+
validateRecords(produceHar.getRequest().version(), records);
890+
} catch (ApiException ex) {
891+
invalidRequestResponses.put(topicPartition,
892+
new ProduceResponse.PartitionResponse(Errors.forException(ex)));
879893
completeOne.run();
880894
return;
881895
}
882-
authorizedRequestInfo.put(topicPartition, records);
883-
completeOne.run();
884-
}, ctx.executor());
896+
final String fullPartitionName = KopTopic.toString(topicPartition, namespacePrefix);
897+
authorize(AclOperation.WRITE, Resource.of(ResourceType.TOPIC, fullPartitionName))
898+
.whenCompleteAsync((isAuthorized, ex) -> {
899+
if (ex != null) {
900+
log.error("Write topic authorize failed, topic - {}. {}",
901+
fullPartitionName, ex.getMessage());
902+
unauthorizedTopicResponsesMap.put(topicPartition,
903+
new ProduceResponse.PartitionResponse(
904+
Errors.TOPIC_AUTHORIZATION_FAILED));
905+
completeOne.run();
906+
return;
907+
}
908+
if (!isAuthorized) {
909+
unauthorizedTopicResponsesMap.put(topicPartition,
910+
new ProduceResponse.PartitionResponse(
911+
Errors.TOPIC_AUTHORIZATION_FAILED));
912+
completeOne.run();
913+
return;
914+
}
915+
authorizedRequestInfo.put(topicPartition, records);
916+
completeOne.run();
917+
}, ctx.executor());
918+
});
885919
});
886-
887-
888920
}
889921

890922
private void validateRecords(short version, MemoryRecords records) {
@@ -1555,7 +1587,7 @@ protected void handleFetchRequest(KafkaHeaderAndRequest fetch,
15551587
}
15561588
String namespacePrefix = currentNamespacePrefix();
15571589
MessageFetchContext.get(this, transactionCoordinator, fetch, resultFuture,
1558-
fetchPurgatory, namespacePrefix).handleFetch();
1590+
fetchPurgatory, namespacePrefix, migrationMetadataManager).handleFetch();
15591591
}
15601592

15611593
@Override

0 commit comments

Comments
 (0)