📋 Overview

🏗️ Architecture

📦 Schemas

CryptoTrade Schema

RiskAlert Schema

PlatformMetadata Schema

🚀 Quick Start

Prerequisites

Installation

💻 Usage

Basic Serialization/Deserialization

importcom.codedstreams.contracts.avro.CryptoTrade;importcom.codedstreams.contracts.avro.RiskAlert;importcom.codedstreams.contracts.AvroSerialization;// Create a trade eventCryptoTradetrade =CryptoTrade.newBuilder()    .setTradeId("trade-12345")    .setUserId("user-001")    .setCustomerId("cust-1001")    .setSymbol("BTC/USDT")    .setBaseAsset("BTC")    .setQuoteAsset("USDT")    .setSide(CryptoTrade.TradeSide.BUY)    .setOrderType(CryptoTrade.OrderType.LIMIT)    .setQuantity(0.5)    .setPrice(45000.0)    .setNotional(22500.0)    .setTimestamp(System.currentTimeMillis())    .setExchange("BINANCE")    .setRiskLevel(CryptoTrade.RiskLevel.LOW)    .build();// Serialize to byte arraybyte[]serializedTrade =AvroSerialization.serializeTrade(trade);// Deserialize from byte arrayCryptoTradedeserializedTrade =AvroSerialization.deserializeTrade(serializedTrade);

Kafka Producer Example

@ComponentpublicclassTradeProducerService {@AutowiredprivateKafkaTemplate<String,byte[]>kafkaTemplate;publicvoidproduceTrade(CryptoTradetrade) {try {byte[]serializedData =AvroSerialization.serializeTrade(trade);kafkaTemplate.send("crypto-trades",trade.getUserId().toString(),serializedData);        }catch (IOExceptione) {thrownewRuntimeException("Failed to serialize trade",e);        }    }}

Kafka Consumer Example

@ComponentpublicclassTradeConsumerService {@KafkaListener(topics ="crypto-trades")publicvoidconsumeTrade(byte[]message) {try {CryptoTradetrade =AvroSerialization.deserializeTrade(message);// Process the trade...processRiskAssessment(trade);        }catch (IOExceptione) {log.error("Failed to deserialize trade",e);        }    }}

🔧 Development

Adding New Schemas

Schema Evolution Best Practices

Example: Adding a New Field

{"name":"newField","type": ["null","string"],"doc":"New field for enhanced risk analysis","default":null}

📊 Schema Registry Integration

// Example configuration for schema registry@ConfigurationpublicclassSchemaRegistryConfig {@BeanpublicSchemaRegistryClientschemaRegistryClient() {returnnewCachedSchemaRegistryClient("http://schema-registry:8081",100        );    }}

🧪 Testing

Unit Testing

publicclassAvroSerializationTest {@TestpublicvoidtestTradeSerializationRoundTrip()throwsIOException {// GivenCryptoTradeoriginalTrade =createTestTrade();// Whenbyte[]serialized =AvroSerialization.serializeTrade(originalTrade);CryptoTradedeserialized =AvroSerialization.deserializeTrade(serialized);// ThenassertEquals(originalTrade.getTradeId(),deserialized.getTradeId());assertEquals(originalTrade.getNotional(),deserialized.getNotional());    }privateCryptoTradecreateTestTrade() {returnCryptoTrade.newBuilder()            .setTradeId("test-trade-001")            .setUserId("test-user-001")            .setCustomerId("test-cust-001")            .setSymbol("BTC/USDT")            .setBaseAsset("BTC")            .setQuoteAsset("USDT")            .setSide(CryptoTrade.TradeSide.BUY)            .setOrderType(CryptoTrade.OrderType.LIMIT)            .setQuantity(1.0)            .setPrice(50000.0)            .setNotional(50000.0)            .setTimestamp(System.currentTimeMillis())            .setExchange("TEST_EXCHANGE")            .setRiskLevel(CryptoTrade.RiskLevel.LOW)            .build();    }}

Integration Testing

@SpringBootTestpublicclassSchemaCompatibilityTest {@TestpublicvoidtestBackwardCompatibility() {// Test that new consumers can read old messages// and old consumers can read new messages    }}

🔒 Security Considerations

📈 Monitoring

Health Metrics

@ComponentpublicclassSchemaHealthIndicatorimplementsHealthIndicator {@OverridepublicHealthhealth() {try {// Validate all schemas can be compiledvalidateSchemas();returnHealth.up()                .withDetail("schemas","healthy")                .withDetail("version","1.0.0")                .build();        }catch (Exceptione) {returnHealth.down(e).build();        }    }}

Logging

@Slf4jpublicclassAvroSerialization {privatestaticfinalLoggerlog =LoggerFactory.getLogger(AvroSerialization.class);publicstaticbyte[]serializeTrade(CryptoTradetrade)throwsIOException {log.debug("Serializing trade: {}",trade.getTradeId());// ... serialization logic    }}

🚀 Deployment

CI/CD Pipeline

name:Build and Deploy Data Contractson:push:branches:[ main ]pull_request:branches:[ main ]jobs:build:runs-on:ubuntu-lateststeps:      -uses:actions/checkout@v3      -name:Set up JDK 11uses:actions/setup-java@v3with:java-version:'11'distribution:'temurin'cache:'maven'      -name:Build and Testrun:mvn clean verify      -name:Deploy to Artifactoryif:github.ref == 'refs/heads/main'run:mvn deployenv:ARTIFACTORY_USER:${{ secrets.ARTIFACTORY_USER }}ARTIFACTORY_PASSWORD:${{ secrets.ARTIFACTORY_PASSWORD }}

Versioning Strategy

🤝 Contributing

Development Workflow

Code Standards

📚 Additional Resources

🆘 Support

📄 License