This repository has been archived by the owner on Jan 24, 2024. It is now read-only.

Request produce implementation #14

Merged

sijie merged 6 commits into streamnative:master from jiazhai:request_produce

Jul 23, 2019

Contributor

jiazhai commented Jul 19, 2019 •

edited by sijie

Master Issue: #4

This is the request produce implementation.
Basic logic:

For passed in PartitionTopic, create related Pulsar persistent topic.
For passed in Kafka Records, write data to related pulsar persistent topic.

Manually tested, kafka client could send data to kop broker:

jiazhai added 3 commits

July 17, 2019 20:14


          fix error of buffer release

46a4259


          success send message out

298bfd8


          fix checkstyle

9ae0e77

jiazhai requested a review from sijie

July 19, 2019 06:42

jiazhai self-assigned this

jiazhai added 2 commits

July 19, 2019 16:59


          fix license check

a13025d


          fix spotbugs:check

sijie suggested changes

View reviewed changes

Member

sijie left a comment

Can you first add a unit to test the request handler first? You don't need setup the server to test. You just need to test handleProduceRequest logic.

src/main/java/io/streamnative/kop/KafkaRequestHandler.java Outdated

                                                       new MetadataResponse(
                                                           allNodes,
                                                           clusterName,
-                                                          MetadataResponse.NO_CONTROLLER_ID,
+,

Member

sijie Jul 19, 2019

what is 0 for?

Contributor Author

jiazhai Jul 23, 2019

Thanks. This is a tmp change. since no controller here, will change it back to NO_CONTROLLER_ID.

src/main/java/io/streamnative/kop/KafkaRequestHandler.java Outdated

+                      for (Map.Entry<TopicPartition, ? extends Records> entry : produceRequest.partitionRecordsOrFail().entrySet()) {
+                          // 1. create PersistTopic
+                          // 3. tracking each record status.

Member

sijie Jul 19, 2019

where is 2?

Contributor Author

jiazhai Jul 23, 2019

Thanks. original thought was to create a producer for each topic at step2 . and then found it is not needed, so removed it. will change this.

src/main/java/io/streamnative/kop/KafkaRequestHandler.java


		// Ignore request.acks() and request.timeout(), which related to kafka replication in this broker.

		Map<TopicPartition, CompletableFuture<PartitionResponse>> responsesFutures = new HashMap<>();

Member

sijie Jul 19, 2019

do you need to use a synchronized hash map?

Contributor Author

jiazhai Jul 23, 2019

Thanks. no need here, requests are handled in a for loop, and no async operation during this hash map put.

src/main/java/io/streamnative/kop/KafkaRequestHandler.java Outdated

+                          TopicName topicName = pulsarTopicName(topicPartition.topic(), topicPartition.partition());
+                          kafkaService.getBrokerService().getTopic(topicName.toString(), true)
+                              .thenApply(Optional::get)

Member

sijie Jul 19, 2019

I don't think you should call Optional::get directly. If the value doesn't exist, it will fail. you have to handle the case that Optional is empty

src/main/java/io/streamnative/kop/KafkaRequestHandler.java Outdated

+                      public void completed(Exception exception, long ledgerId, long entryId) {
+                          if (exception != null) {
+                              log.debug("Failed write entry: {}, entryId: {}, sequenceId: {}. and triggered send callback.",

Member

sijie Jul 19, 2019

why this is debug not error?

Contributor Author

jiazhai Jul 23, 2019

will change to error.

src/main/java/io/streamnative/kop/KafkaRequestHandler.java

+                      List<CompletableFuture<Long>> futures = Collections
+                          .synchronizedList(Lists.newArrayListWithExpectedSize(size.get()));
+                      records.records().forEach(record -> {

Member

sijie Jul 19, 2019

Can we batch multiple records into one message batch?

Contributor Author

jiazhai Jul 23, 2019 •

edited

Added an issue to handle this. #16

sijie added area/protocol type/feature labels

sijie added this to the 0.0.1 milestone

sijie mentioned this pull request

Kafka-on-Pulsar MVP #4

Closed

12 tasks


          change following comments

a09a805

jiazhai mentioned this pull request

Add test framework to be able to run unit test. #17

Merged

Contributor Author

jiazhai commented Jul 23, 2019

@sijie updated, and test framework is added in PR #17 .

sijie approved these changes

View reviewed changes

sijie merged commit ee0ffb2 into streamnative:master

sijie deleted the request_produce branch

July 23, 2019 13:59

sijie pushed a commit that referenced this pull request


          Add test framework to be able to run unit test. (#17)

daa19e1

This PR is based on 
[Produce request](a09a805), which is in PR #14, please review and merge that first.

changes:
- change to sn pulsar version, so we could call getter/setter for parent class in `KafkaBrokerService` and `KafkaService`. Or it is not able to do the mock in test framework.
- add test framework, 
- add a simple produce request test, and pass write test.

jiazhai modified the milestones: 0.0.1, 0.1.0

michaeljmarshall pushed a commit to michaeljmarshall/kop that referenced this pull request


          Feature: auto-register AVRO schema id while serving Fetch (kafkaApply…

cb659ab

…AvroSchemaOnDecode) (streamnative#14)

Demogorgon314 pushed a commit to Demogorgon314/kop that referenced this pull request


          Support setting up a KDC server in tests (streamnative#14)

d977dab

Master issue: streamnative/ksn#7

### Motivation

A KDC server is required to run a Kerberos authentication. Before adding
the implementation of Kerberos authentication, this PR aimes at setting
up a KDC server along with the KoP service.

### Modification

- Add a `MiniKdc` implementation based on the [Hadoop
MiniKdc](https://github.com/apache/hadoop/blob/trunk/hadoop-common-project/hadoop-minikdc/src/main/java/org/apache/hadoop/minikdc/MiniKdc.java)
- Migrate some classes like `SaslSetup` and `JaasUtils` to save some
JAAS configuration files.
- Add a trivial `GssapiAuthenticationTest` to verify the KDC server has
started.

### Documentation

Check the box below.

Need to update docs? 

- [ ] `doc-required` 
  
  (If you need help on updating docs, create a doc issue)
  
- [x] `no-need-doc` 
  
  (Please explain why)
  
- [ ] `doc` 
  
  (If this PR contains doc changes)

Demogorgon314 pushed a commit to Demogorgon314/kop that referenced this pull request


          Support setting up a KDC server in tests (streamnative#14)

6a86b15

Master issue: streamnative/ksn#7

### Motivation

A KDC server is required to run a Kerberos authentication. Before adding
the implementation of Kerberos authentication, this PR aimes at setting
up a KDC server along with the KoP service.

### Modification

- Add a `MiniKdc` implementation based on the [Hadoop
MiniKdc](https://github.com/apache/hadoop/blob/trunk/hadoop-common-project/hadoop-minikdc/src/main/java/org/apache/hadoop/minikdc/MiniKdc.java)
- Migrate some classes like `SaslSetup` and `JaasUtils` to save some
JAAS configuration files.
- Add a trivial `GssapiAuthenticationTest` to verify the KDC server has
started.

### Documentation

Check the box below.

Need to update docs?

- [ ] `doc-required`

  (If you need help on updating docs, create a doc issue)

- [x] `no-need-doc`

  (Please explain why)

- [ ] `doc`

  (If this PR contains doc changes)

(cherry picked from commit d977dab)

Sign up for free to subscribe to this conversation on GitHub. Already have an account? Sign in.