Skip to content

Commit a8ecb36

Browse files
author
Walter Schultz
committed
Added a check for sft in the kds, if it isn't on the classpath
1 parent 70bb161 commit a8ecb36

9 files changed

Lines changed: 94 additions & 34 deletions

File tree

‎geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/pom.xml‎

Lines changed: 27 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,6 @@
1313
<name>GeoMesa Kafka Spring Cloud Stream Binder</name>
1414

1515
<properties>
16-
<maven.compiler.source>11</maven.compiler.source>
17-
<maven.compiler.target>11</maven.compiler.target>
1816
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
1917
<spring-cloud.version>2021.0.9</spring-cloud.version>
2018
<spring-boot.version>2.7.18</spring-boot.version>
@@ -32,7 +30,14 @@
3230
<dependency>
3331
<groupId>org.springframework.boot</groupId>
3432
<artifactId>spring-boot-starter-log4j2</artifactId>
33+
<exclusions>
34+
<exclusion>
35+
<groupId>org.apache.logging.log4j</groupId>
36+
<artifactId>log4j-slf4j-impl</artifactId>
37+
</exclusion>
38+
</exclusions>
3539
</dependency>
40+
3641
<!-- NOTE: needed to work with GeoMesa -->
3742
<dependency>
3843
<groupId>org.apache.curator</groupId>
@@ -62,11 +67,29 @@
6267
<groupId>org.apache.logging.log4j</groupId>
6368
<artifactId>log4j-to-slf4j</artifactId>
6469
</exclusion>
70+
<exclusion>
71+
<groupId>ch.qos.logback</groupId>
72+
<artifactId>logback-classic</artifactId>
73+
</exclusion>
74+
<exclusion>
75+
<groupId>org.junit.jupiter</groupId>
76+
<artifactId>junit-jupiter</artifactId>
77+
</exclusion>
6578
</exclusions>
6679
</dependency>
6780
<dependency>
68-
<groupId>org.junit.vintage</groupId>
69-
<artifactId>junit-vintage-engine</artifactId>
81+
<groupId>junit</groupId>
82+
<artifactId>junit</artifactId>
83+
<scope>test</scope>
84+
</dependency>
85+
<!-- <dependency>-->
86+
<!-- <groupId>org.junit.vintage</groupId>-->
87+
<!-- <artifactId>junit-vintage-engine</artifactId>-->
88+
<!-- <scope>test</scope>-->
89+
<!-- </dependency>-->
90+
<dependency>
91+
<groupId>org.slf4j</groupId>
92+
<artifactId>slf4j-reload4j</artifactId>
7093
<scope>test</scope>
7194
</dependency>
7295
</dependencies>

geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/spring/binder/kafka/datastore/KafkaDatastoreBinderConfiguration.java renamed to geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/kafka/spring/binder/KafkaDatastoreBinderConfiguration.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,11 +6,11 @@
66
* http://www.opensource.org/licenses/apache2.0.php.
77
***********************************************************************/
88

9-
package org.locationtech.geomesa.spring.binder.kafka.datastore;
9+
package org.locationtech.geomesa.kafka.spring.binder;
1010

1111
import org.geotools.api.data.DataStore;
1212
import org.geotools.api.data.DataStoreFinder;
13-
import org.locationtech.geomesa.spring.binder.kafka.datastore.converters.SimpleFeatureConverter;
13+
import org.locationtech.geomesa.kafka.spring.binder.converters.SimpleFeatureConverter;
1414
import org.slf4j.Logger;
1515
import org.slf4j.LoggerFactory;
1616
import org.springframework.beans.factory.annotation.Autowired;

geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/spring/binder/kafka/datastore/KafkaDatastoreBinderConfigurationProperties.java renamed to geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/kafka/spring/binder/KafkaDatastoreBinderConfigurationProperties.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
* http://www.opensource.org/licenses/apache2.0.php.
77
***********************************************************************/
88

9-
package org.locationtech.geomesa.spring.binder.kafka.datastore;
9+
package org.locationtech.geomesa.kafka.spring.binder;
1010

1111
import org.springframework.boot.context.properties.ConfigurationProperties;
1212

@@ -15,7 +15,7 @@
1515
import java.util.Map;
1616

1717
@ConfigurationProperties(prefix = "spring.cloud.stream.kafka-datastore")
18-
public class KafkaDatastoreBinderConfigurationProperties {
18+
public class KafkaDatastoreBinderConfigurationProperties {
1919
public Map<String, ? extends Serializable> binder = new HashMap<>();
2020

2121
public Map<String, ? extends Serializable> getBinder() {

geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/spring/binder/kafka/datastore/KafkaDatastoreBinderProvisioner.java renamed to geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/kafka/spring/binder/KafkaDatastoreBinderProvisioner.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
* http://www.opensource.org/licenses/apache2.0.php.
77
***********************************************************************/
88

9-
package org.locationtech.geomesa.spring.binder.kafka.datastore;
9+
package org.locationtech.geomesa.kafka.spring.binder;
1010

1111
import org.springframework.cloud.stream.binder.ConsumerProperties;
1212
import org.springframework.cloud.stream.binder.ProducerProperties;

geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/spring/binder/kafka/datastore/KafkaDatastoreMessageBinder.java renamed to geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/kafka/spring/binder/KafkaDatastoreMessageBinder.java

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
* http://www.opensource.org/licenses/apache2.0.php.
77
***********************************************************************/
88

9-
package org.locationtech.geomesa.spring.binder.kafka.datastore;
9+
package org.locationtech.geomesa.kafka.spring.binder;
1010

1111
import org.geotools.api.data.DataStore;
1212
import org.geotools.api.data.FeatureWriter;
@@ -69,7 +69,12 @@ protected MessageHandler createProducerMessageHandler(
6969
if (sft.isDefined()) {
7070
ds.createSchema(sft.get());
7171
} else {
72-
logger.warn("Could not find a local version of {}, hoping the KDS is already defined...", sfName);
72+
try {
73+
ds.getSchema(sfName);
74+
logger.debug("There is no local schema for {}, but we found it in the kds", sfName);
75+
} catch (IOException e) {
76+
logger.error("There is no sft schema {} in the kds {} or locally", sfName, ds.getInfo().getDescription(), e);
77+
}
7378
}
7479

7580
if (message.getPayload() instanceof SimpleFeature) {

geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/spring/binder/kafka/datastore/KafkaDatastoreMessageProducer.java renamed to geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/kafka/spring/binder/KafkaDatastoreMessageProducer.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
* http://www.opensource.org/licenses/apache2.0.php.
77
***********************************************************************/
88

9-
package org.locationtech.geomesa.spring.binder.kafka.datastore;
9+
package org.locationtech.geomesa.kafka.spring.binder;
1010

1111

1212
import org.geotools.api.data.DataStore;

geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/spring/binder/kafka/datastore/converters/SimpleFeatureConverter.java renamed to geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/java/org/locationtech/geomesa/kafka/spring/binder/converters/SimpleFeatureConverter.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
* http://www.opensource.org/licenses/apache2.0.php.
77
***********************************************************************/
88

9-
package org.locationtech.geomesa.spring.binder.kafka.datastore.converters;
9+
package org.locationtech.geomesa.kafka.spring.binder.converters;
1010

1111
import org.locationtech.geomesa.kafka.utils.KafkaFeatureEvent;
1212
import org.geotools.api.feature.simple.SimpleFeature;

‎geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/main/resources/META-INF/spring.binders‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,2 @@
11
kafka-datastore:\
2-
com.ccri.geomesa.binder.kafka.datastore.KafkaDatastoreBinderConfiguration
2+
org.locationtech.geomesa.kafka.spring.binder.KafkaDatastoreBinderConfiguration

geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/test/java/org/locationtech/geomesa/spring/binder/kafka/datastore/KafkaDatastoreMessageBinderTest.java renamed to geomesa-kafka/geomesa-kafka-spring-cloud-stream-binder/src/test/java/org/locationtech/geomesa/kafka/spring/binder/KafkaDatastoreMessageBinderTest.java

Lines changed: 52 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
* http://www.opensource.org/licenses/apache2.0.php.
77
***********************************************************************/
88

9-
package org.locationtech.geomesa.spring.binder.kafka.datastore;
9+
package org.locationtech.geomesa.kafka.spring.binder;
1010

1111
import org.geotools.api.data.DataStore;
1212
import org.geotools.api.data.FeatureWriter;
@@ -15,12 +15,10 @@
1515
import org.geotools.api.feature.simple.SimpleFeatureType;
1616
import org.geotools.api.filter.Filter;
1717
import org.geotools.feature.simple.SimpleFeatureBuilder;
18-
import org.junit.jupiter.api.Test;
19-
import org.junit.jupiter.api.extension.ExtendWith;
18+
import org.junit.Before;
19+
import org.junit.Test;
2020
import org.locationtech.geomesa.kafka.utils.KafkaFeatureEvent;
2121
import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypeLoader;
22-
import org.mockito.Mock;
23-
import org.mockito.junit.jupiter.MockitoExtension;
2422
import org.springframework.cloud.stream.binder.ProducerProperties;
2523
import org.springframework.cloud.stream.provisioning.ProducerDestination;
2624
import org.springframework.messaging.Message;
@@ -31,29 +29,33 @@
3129
import java.io.IOException;
3230
import java.time.Instant;
3331
import java.util.ArrayList;
32+
import java.util.Date;
3433
import java.util.function.Supplier;
3534

3635
import static org.assertj.core.api.Assertions.assertThat;
37-
import static org.junit.jupiter.api.Assertions.*;
3836
import static org.mockito.ArgumentMatchers.any;
3937
import static org.mockito.Mockito.*;
4038

41-
@ExtendWith(MockitoExtension.class)
42-
class KafkaDatastoreMessageBinderTest {
39+
public class KafkaDatastoreMessageBinderTest {
4340

44-
@Mock
4541
DataStore ds;
46-
@Mock
4742
FeatureWriter featureWriter;
48-
@Mock
4943
SimpleFeatureStore simpleFeatureStore;
50-
@Mock
5144
SubscribableChannel errorChannel;
52-
53-
Supplier<DataStore> dsFactory = () -> ds;
45+
Supplier<DataStore> dsFactory;
46+
47+
@Before
48+
public void init() {
49+
ds = mock(DataStore.class);
50+
featureWriter = mock(FeatureWriter.class);
51+
simpleFeatureStore = mock(SimpleFeatureStore.class);
52+
errorChannel = mock(SubscribableChannel.class);
53+
dsFactory = () -> ds;
54+
}
5455

5556
@Test
5657
public void producerMessageHandler_canWriteSft() throws IOException {
58+
5759
SimpleFeatureType sft = SimpleFeatureTypeLoader.sftForName("observation").get();
5860
SimpleFeature writeableFeature = SimpleFeatureBuilder.build(sft, new ArrayList<>(), "id0");
5961
doReturn(featureWriter).when(ds).getFeatureWriterAppend(any(), any());
@@ -70,19 +72,20 @@ public void producerMessageHandler_canWriteSft() throws IOException {
7072

7173

7274
SimpleFeature simpleFeature = SimpleFeatureBuilder.build(sft, new ArrayList<>(), "id1");
73-
simpleFeature.setAttribute("mmsi", 123456);
75+
simpleFeature.setAttribute("id", "123456");
7476
Message<?> message = MessageBuilder.withPayload(simpleFeature)
7577
.setHeader("featureType", "application/simple-feature")
7678
.build();
7779

7880
handler.handleMessage(message);
7981

8082
verify(featureWriter).write();
81-
assertThat(writeableFeature.getAttribute("mmsi")).isEqualTo(123456);
83+
assertThat(writeableFeature.getAttribute("id")).isEqualTo("123456");
8284
}
8385

8486
@Test
8587
public void producerMessageHandler_canWriteKafkaFeatureEventChanged() throws IOException {
88+
var now = Instant.now();
8689
SimpleFeatureType sft = SimpleFeatureTypeLoader.sftForName("observation").get();
8790
SimpleFeature writeableFeature = SimpleFeatureBuilder.build(sft, new ArrayList<>(), "id0");
8891
doReturn(featureWriter).when(ds).getFeatureWriterAppend(any(), any());
@@ -99,8 +102,8 @@ public void producerMessageHandler_canWriteKafkaFeatureEventChanged() throws IOE
99102

100103

101104
SimpleFeature simpleFeature = SimpleFeatureBuilder.build(sft, new ArrayList<>(), "id1");
102-
simpleFeature.setAttribute("mmsi", 123456);
103-
simpleFeature.setAttribute("elevation", 100);
105+
simpleFeature.setAttribute("id", "123456");
106+
simpleFeature.setAttribute("dtg", now);
104107
KafkaFeatureEvent changed = new KafkaFeatureEvent.KafkaFeatureChanged("test", simpleFeature, Instant.now().getEpochSecond());
105108
Message<?> message = MessageBuilder.withPayload(changed)
106109
.setHeader("featureType", "application/kafka-feature-event")
@@ -109,8 +112,8 @@ public void producerMessageHandler_canWriteKafkaFeatureEventChanged() throws IOE
109112
handler.handleMessage(message);
110113

111114
verify(featureWriter).write();
112-
assertThat(writeableFeature.getAttribute("mmsi")).isEqualTo(123456);
113-
assertThat(writeableFeature.getAttribute("elevation")).isEqualTo(100f);
115+
assertThat(writeableFeature.getAttribute("id")).isEqualTo("123456");
116+
assertThat(writeableFeature.getAttribute("dtg")).isEqualTo(Date.from(now));
114117
}
115118

116119
@Test
@@ -165,4 +168,33 @@ public void producerMessageHandler_canWriteKafkaFeatureEventClear() throws IOExc
165168
verify(simpleFeatureStore).removeFeatures(Filter.INCLUDE);
166169
}
167170

171+
@Test
172+
public void producerMessageHandler_loadSftFromClasspath() throws IOException {
173+
SimpleFeatureType sft = SimpleFeatureTypeLoader.sftForName("observation").get();
174+
SimpleFeature writeableFeature = SimpleFeatureBuilder.build(sft, new ArrayList<>(), "id0");
175+
doReturn(featureWriter).when(ds).getFeatureWriterAppend(any(), any());
176+
doReturn(writeableFeature).when(featureWriter).next();
177+
178+
KafkaDatastoreBinderProvisioner provisioningProvider = new KafkaDatastoreBinderProvisioner();
179+
provisioningProvider = new KafkaDatastoreBinderProvisioner();
180+
KafkaDatastoreMessageBinder messageBinder = new KafkaDatastoreMessageBinder(new String[]{}, provisioningProvider, dsFactory);
181+
182+
ProducerProperties producerProperties = new ProducerProperties();
183+
ProducerDestination destination = provisioningProvider.provisionProducerDestination("observation", producerProperties);
184+
185+
// Doesn't throw an error
186+
MessageHandler handler = messageBinder.createProducerMessageHandler(destination, producerProperties, errorChannel);
187+
188+
SimpleFeature simpleFeature = SimpleFeatureBuilder.build(sft, new ArrayList<>(), "id1");
189+
simpleFeature.setAttribute("id", "123456");
190+
KafkaFeatureEvent changed = new KafkaFeatureEvent.KafkaFeatureChanged("test", simpleFeature, Instant.now().getEpochSecond());
191+
Message<?> message = MessageBuilder.withPayload(changed)
192+
.setHeader("featureType", "application/kafka-feature-event")
193+
.build();
194+
195+
handler.handleMessage(message);
196+
197+
verify(ds).createSchema(any());
198+
}
199+
168200
}

0 commit comments

Comments
 (0)