Pre create a dummy topic and subscribe to it during first POST call for
posting a message to create __consumer_offsets topic
Issue-ID: DMAAP-896
Change-Id: I11f3f9b8764232bc7d4e9bb270d5d73fc280cb80
Signed-off-by: sunil unnava <sunil.unnava@att.com>
@Autowired
private DMaaPErrorMessages errorMessages;
@Autowired
private DMaaPErrorMessages errorMessages;
+
+ private boolean isOffsetTopicCreated=false;
/**
* This method is used to consume messages.Taking three parameter
/**
* This method is used to consume messages.Taking three parameter
public void pushEvents(@PathParam("topic") String topic, InputStream msg,
@QueryParam("partitionKey") String partitionKey) throws CambriaApiException {
log.info("Publishing message to topic " + topic);
public void pushEvents(@PathParam("topic") String topic, InputStream msg,
@QueryParam("partitionKey") String partitionKey) throws CambriaApiException {
log.info("Publishing message to topic " + topic);
+
+ if(!isOffsetTopicCreated){
+ preCreateOffsetTopic(msg);
+ }
try {
eventsService.pushEvents(getDmaapContext(), topic, msg, partitionKey, null);
}
try {
eventsService.pushEvents(getDmaapContext(), topic, msg, partitionKey, null);
}
@QueryParam("partitionKey") String partitionKey) throws CambriaApiException {
// log.info("Publishing message with transaction id for topic " + topic
// );
@QueryParam("partitionKey") String partitionKey) throws CambriaApiException {
// log.info("Publishing message with transaction id for topic " + topic
// );
+
+ if(!isOffsetTopicCreated){
+ preCreateOffsetTopic(request.getInputStream());
+ }
+
eventsService.pushEvents(getDmaapContext(), topic, request.getInputStream(),
partitionKey,
Utils.getFormattedDate(new Date()));
eventsService.pushEvents(getDmaapContext(), topic, request.getInputStream(),
partitionKey,
Utils.getFormattedDate(new Date()));
+
+ private void preCreateOffsetTopic(InputStream msg) {
+
+ try {
+ eventsService.pushEvents(getDmaapContext(), "DUMMY_TOPIC", msg, null, null);
+ eventsService.getEvents(getDmaapContext(), "DUMMY_TOPIC", "CG1", "C1");
+ isOffsetTopicCreated = true;
+ } catch (CambriaApiException | ConfigDbException | AccessDeniedException | TopicExistsException | IOException
+ | missingReqdSetting | UnavailableException e) {
+ log.error("Error while creating the dummy topic", e);
+ }
+
+ }
}
\ No newline at end of file
}
\ No newline at end of file