Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
I
inventory-service
Project
Project
Details
Activity
Releases
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
Ascend
inventory-service
Commits
801065ba
Commit
801065ba
authored
May 19, 2021
by
Ben Anderson
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Deployment 58
parent
5061bffd
Changes
2
Show whitespace changes
Inline
Side-by-side
Showing
2 changed files
with
68 additions
and
68 deletions
+68
-68
deployment.yaml
deployment.yaml
+1
-1
KafkaReceiverConfig.java
...m/ascend/inventory/configuration/KafkaReceiverConfig.java
+67
-67
No files found.
deployment.yaml
View file @
801065ba
...
@@ -14,7 +14,7 @@ spec:
...
@@ -14,7 +14,7 @@ spec:
spec
:
spec
:
containers
:
containers
:
-
name
:
afp-prods-container
-
name
:
afp-prods-container
image
:
nexus.mynisum.com/afp-prods:5
7
image
:
nexus.mynisum.com/afp-prods:5
8
imagePullPolicy
:
Always
imagePullPolicy
:
Always
ports
:
ports
:
-
containerPort
:
8083
-
containerPort
:
8083
...
...
src/main/java/com/nisum/ascend/inventory/configuration/KafkaReceiverConfig.java
View file @
801065ba
//package com.nisum.ascend.inventory.configuration;
package
com
.
nisum
.
ascend
.
inventory
.
configuration
;
//
//
//import com.nisum.ascend.inventory.dto.Order;
import
com.nisum.ascend.inventory.dto.Order
;
//import com.nisum.ascend.inventory.dto.WareHouseOrder;
import
com.nisum.ascend.inventory.dto.WareHouseOrder
;
//import lombok.extern.slf4j.Slf4j;
import
lombok.extern.slf4j.Slf4j
;
//import org.apache.kafka.clients.consumer.ConsumerConfig;
import
org.apache.kafka.clients.consumer.ConsumerConfig
;
//import org.apache.kafka.clients.producer.ProducerConfig;
import
org.apache.kafka.clients.producer.ProducerConfig
;
//import org.apache.kafka.common.serialization.StringDeserializer;
import
org.apache.kafka.common.serialization.StringDeserializer
;
//import org.springframework.beans.factory.annotation.Value;
import
org.springframework.beans.factory.annotation.Value
;
//import org.springframework.context.annotation.Bean;
import
org.springframework.context.annotation.Bean
;
//import org.springframework.context.annotation.Configuration;
import
org.springframework.context.annotation.Configuration
;
//import org.springframework.kafka.support.serializer.JsonDeserializer;
import
org.springframework.kafka.support.serializer.JsonDeserializer
;
//import reactor.kafka.receiver.KafkaReceiver;
import
reactor.kafka.receiver.KafkaReceiver
;
//import reactor.kafka.receiver.ReceiverOptions;
import
reactor.kafka.receiver.ReceiverOptions
;
//
//import java.util.Collection;
import
java.util.Collection
;
//import java.util.Collections;
import
java.util.Collections
;
//import java.util.HashMap;
import
java.util.HashMap
;
//import java.util.Map;
import
java.util.Map
;
//
//@Configuration
@Configuration
//@Slf4j
@Slf4j
//public class KafkaReceiverConfig {
public
class
KafkaReceiverConfig
{
//
// @Value("${kafka.consumer.bootstrap-servers}")
@Value
(
"${kafka.consumer.bootstrap-servers}"
)
// private String bootstrapServers;
private
String
bootstrapServers
;
// @Value("${kafka.consumer.group-id}")
@Value
(
"${kafka.consumer.group-id}"
)
// private String groupId;
private
String
groupId
;
//
// @Bean("kafkaWarehouseOrderReceiver")
@Bean
(
"kafkaWarehouseOrderReceiver"
)
// public KafkaReceiver<String, String> kafkaWarehouseOrderEventReceiver(
public
KafkaReceiver
<
String
,
String
>
kafkaWarehouseOrderEventReceiver
(
// @Value("${kafka.WAREHOUSETOPIC.input}") String posLogTopic) {
@Value
(
"${kafka.WAREHOUSETOPIC.input}"
)
String
posLogTopic
)
{
// ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(WarehouseOrderEventReceiverConfig());
ReceiverOptions
<
String
,
String
>
receiverOptions
=
ReceiverOptions
.
create
(
WarehouseOrderEventReceiverConfig
());
// receiverOptions.maxCommitAttempts(3);
receiverOptions
.
maxCommitAttempts
(
3
);
// return KafkaReceiver.create(receiverOptions.addAssignListener(Collection::iterator)
return
KafkaReceiver
.
create
(
receiverOptions
.
addAssignListener
(
Collection:
:
iterator
)
// .subscription(Collections.singleton(posLogTopic)));
.
subscription
(
Collections
.
singleton
(
posLogTopic
)));
// }
}
//
// private Map<String, Object> WarehouseOrderEventReceiverConfig() {
private
Map
<
String
,
Object
>
WarehouseOrderEventReceiverConfig
()
{
// Map<String, Object> config = new HashMap<>();
Map
<
String
,
Object
>
config
=
new
HashMap
<>();
// config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
config
.
put
(
ConsumerConfig
.
GROUP_ID_CONFIG
,
groupId
);
// config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
config
.
put
(
ProducerConfig
.
BOOTSTRAP_SERVERS_CONFIG
,
bootstrapServers
);
// config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config
.
put
(
ConsumerConfig
.
KEY_DESERIALIZER_CLASS_CONFIG
,
StringDeserializer
.
class
);
// config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config
.
put
(
ConsumerConfig
.
VALUE_DESERIALIZER_CLASS_CONFIG
,
StringDeserializer
.
class
);
// return config;
return
config
;
// }
}
//
// @Bean("kafkaOrderReceiver")
@Bean
(
"kafkaOrderReceiver"
)
// public KafkaReceiver<String, String> kafkaOrderEventReceiver(
public
KafkaReceiver
<
String
,
String
>
kafkaOrderEventReceiver
(
// @Value("${kafka.ORDERTOPIC.input}") String posLogTopic) {
@Value
(
"${kafka.ORDERTOPIC.input}"
)
String
posLogTopic
)
{
// ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(OrderEventReceiverConfig());
ReceiverOptions
<
String
,
String
>
receiverOptions
=
ReceiverOptions
.
create
(
OrderEventReceiverConfig
());
// receiverOptions.maxCommitAttempts(3);
receiverOptions
.
maxCommitAttempts
(
3
);
// return KafkaReceiver.create(receiverOptions.addAssignListener(Collection::iterator)
return
KafkaReceiver
.
create
(
receiverOptions
.
addAssignListener
(
Collection:
:
iterator
)
// .subscription(Collections.singleton(posLogTopic)));
.
subscription
(
Collections
.
singleton
(
posLogTopic
)));
// }
}
//
// private Map<String, Object> OrderEventReceiverConfig() {
private
Map
<
String
,
Object
>
OrderEventReceiverConfig
()
{
// Map<String, Object> config = new HashMap<>();
Map
<
String
,
Object
>
config
=
new
HashMap
<>();
// config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
config
.
put
(
ConsumerConfig
.
GROUP_ID_CONFIG
,
groupId
);
// config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
config
.
put
(
ProducerConfig
.
BOOTSTRAP_SERVERS_CONFIG
,
bootstrapServers
);
// config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config
.
put
(
ConsumerConfig
.
KEY_DESERIALIZER_CLASS_CONFIG
,
StringDeserializer
.
class
);
// config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config
.
put
(
ConsumerConfig
.
VALUE_DESERIALIZER_CLASS_CONFIG
,
StringDeserializer
.
class
);
// return config;
return
config
;
// }
}
//
//}
}
\ No newline at end of file
\ No newline at end of file
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment